lora_patch
The part of a LoRA adapter that is not a low-rank product.
A LoRA states a weight delta as B @ A. That form needs two things to be true: the
base checkpoint must already contain the parameter, and the delta must actually be low
rank. Distilled video checkpoints break both often enough that dropping whatever does
not fit silently loses real signal.
Two payload kinds cover the gap. The weight/bias spellings follow conventions used by
ComfyUI; the *_param spellings extend them to standalone FastVideo parameters:
<module>.diff / <module>.diff_b / <parameter>.diff_param
An exact additive delta for a parameter the base model has. Used where a rank-r
factorization buys nothing or cannot be formed at all -- RMSNorm vectors, biases,
standalone parameters such as scale_shift_table, and matrices whose smaller
dimension is already at or below the rank that would be chosen. Factoring a
length-n vector into rank r costs r(1 + n) > n.
<module>.set_weight / <parameter>.set_param
A whole parameter the base model does not carry, so no delta is expressible. MiniMax
H3's VSA to_gate_compress is the case that motivated this: it exists only under
the sparse-attention backend, and :func:load_model_from_full_model_state_dict
otherwise zero-initializes it, which is exactly the "gate contributes nothing"
state a VSA-distilled student was trained away from.
Both kinds are applied to the unsharded tensor while the checkpoint is streaming in,
so FSDP and tensor-parallel placement come from the surrounding loader instead of being
reimplemented here. That ordering is not a preference: maybe_load_fsdp_model shards
the model on the meta device before any weight is read, so there is no earlier window,
and patching afterwards would mean gathering and redistributing every affected
parameter one at a time.
Because the adapter has to be known when the transformer loads, this path applies the
adapter a pipeline is constructed with. Swapping to a different adapter later still goes
through :meth:LoRAPipeline.set_lora_adapter, which handles the low-rank half only.
fastvideo.models.loader.lora_patch.init_logger
init_logger(name: str) -> _FastvideoLoggerThe main purpose of this function is to ensure that loggers are retrieved in such a way that we can be sure the root fastvideo logger has already been configured.
fastvideo.models.loader.lora_patch.normalize_lora_key
normalize_lora_key(name: str) -> str | NoneRewrite an adapter key to the <module>.lora_A|lora_B|lora_alpha spelling.
Returns None for keys the low-rank merge path deliberately does not handle --
the dense payload above, which lands during checkpoint load, and bookkeeping like
.dora_scale. Callers use that to tell "not mine" apart from "mine and
unmatched", which is the difference between a quiet skip and a warning.
Source
def normalize_lora_key(name: str) -> str | None:
"""Rewrite an adapter key to the ``<module>.lora_A|lora_B|lora_alpha`` spelling.
Returns ``None`` for keys the low-rank merge path deliberately does not handle --
the dense payload above, which lands during checkpoint load, and bookkeeping like
``.dora_scale``. Callers use that to tell "not mine" apart from "mine and
unmatched", which is the difference between a quiet skip and a warning.
"""
if name.endswith(tuple(ADDITIVE_SUFFIXES) + tuple(REPLACEMENT_SUFFIXES)) or name.endswith(".dora_scale"):
return None
name = name.replace("diffusion_model.", "")
name = _ADAPTER_NAME_INFIX.sub(r".\1.weight", name)
for alias, canonical in _LOW_RANK_ALIASES:
name = name.replace(alias + ".", canonical + ".")
if name.endswith(alias):
name = name[:-len(alias)] + canonical
if name.endswith(".alpha"):
name = name[:-len(".alpha")] + ".lora_alpha"
return name</code></pre></details>
fastvideo.models.loader.lora_patch.DenseLoRAPatch
DenseLoRAPatch(files: list[str], additive: dict[str, tuple[str, str]], replacement: dict[str, tuple[str, str]], strength: float = 1.0)
Adapter keys that address a whole parameter rather than a factor of one.
Construction only reads the safetensors headers, so the tensors themselves stay on
disk until the loader asks for one. That keeps peak host memory at a single
parameter even for adapters whose dense half is several GiB, which the VSA gates
alone are.
Source
class DenseLoRAPatch:
"""Adapter keys that address a whole parameter rather than a factor of one.
Construction only reads the safetensors headers, so the tensors themselves stay on
disk until the loader asks for one. That keeps peak host memory at a single
parameter even for adapters whose dense half is several GiB, which the VSA gates
alone are.
"""
def __init__(self, files: list[str], additive: dict[str, tuple[str, str]],
replacement: dict[str, tuple[str, str]], strength: float = 1.0) -> None:
if not math.isfinite(strength):
raise ValueError(f"LoRA strength must be finite, got {strength}")
self._files = files
# target parameter name -> (file, adapter key)
self._additive = additive
self._replacement = replacement
self._strength = float(strength)
self._applied: set[str] = set()
@classmethod
def from_adapter(
cls,
lora_path: str | None,
param_names_mapping: Callable[[str], tuple[str, Any, Any]] | None = None,
*,
lora_param_names_mapping: Callable[[str], tuple[str, Any, Any]] | None = None,
strength: float = 1.0,
) -> DenseLoRAPatch | None:
"""Build a patch from an adapter, or ``None`` when it carries no dense payload.
``lora_param_names_mapping`` first translates adapter-specific official names
into the published checkpoint layout. ``param_names_mapping`` then resolves that
layout into the model's parameter names, matching the low-rank loader's order.
"""
if not lora_path:
return None
# Deferred: fastvideo.utils pulls in enough of the package that importing it at
# module scope would make this loader helper part of an import cycle.
from fastvideo.utils import maybe_download_lora
files = _adapter_files(maybe_download_lora(lora_path))
if not files:
logger.warning("LoRA path %s holds no safetensors file; no dense patch applied", lora_path)
return None
additive: dict[str, tuple[str, str]] = {}
replacement: dict[str, tuple[str, str]] = {}
for path in files:
with safe_open(path, framework="pt") as handle:
for key in handle.keys():
resolved = _resolve(key, lora_param_names_mapping, param_names_mapping)
if resolved is None:
continue
target, kind = resolved
table = additive if kind == "add" else replacement
if target in table:
raise ValueError(f"LoRA adapter {lora_path} maps two keys onto parameter {target}; "
f"the second is {key}")
table[target] = (path, key)
if not additive and not replacement:
return None
logger.info(
"LoRA adapter %s carries a dense payload: %d additive, %d replacement parameters",
lora_path, len(additive), len(replacement))
return cls(files, additive, replacement, strength)
def apply_to(self, param_name: str, tensor: torch.Tensor) -> torch.Tensor:
"""Add this parameter's ``.diff``/``.diff_b`` delta, if the adapter has one.
The sum is taken in float32 whatever the operands are: both sides arrive in
bfloat16, whose 8-bit significand would quantize away a delta three orders of
magnitude below the base weight. The caller casts back to the target dtype.
"""
entry = self._additive.get(param_name)
if entry is None:
return tensor
delta = self._read(entry)
if delta.shape != tensor.shape:
raise ValueError(f"LoRA diff for {param_name} has shape {tuple(delta.shape)}, "
f"but the parameter is {tuple(tensor.shape)}")
self._applied.add(param_name)
# Safetensors returns the adapter delta on CPU, while inference can stream the
# base checkpoint directly onto its target GPU when CPU offload is disabled.
delta = delta.to(device=tensor.device, dtype=torch.float32)
return tensor.to(torch.float32) + delta * self._strength
def provides(self, param_name: str) -> bool:
"""Whether the adapter carries this parameter whole, without reading it."""
return param_name in self._replacement
@property
def replacement_parameters(self) -> frozenset[str]:
"""Names of the parameters this adapter supplies outright.
Lets a caller decide what the adapter needs from the runtime -- an H3 adapter
carrying ``to_gate_compress`` only works under the VSA backend -- without
reading any tensor or guessing from the file name.
"""
return frozenset(self._replacement)
def replacement_for(self, param_name: str) -> torch.Tensor | None:
"""The adapter's whole-tensor value for a parameter, or ``None``."""
entry = self._replacement.get(param_name)
if entry is None:
return None
self._applied.add(param_name)
# Replacement payloads address parameters absent from the base checkpoint. The
# loader initializes those parameters to zero, so scaling the supplied value is
# the same interpolation contract as ``base + strength * delta``.
return self._read(entry).to(torch.float32) * self._strength
def report_unapplied(self) -> None:
"""Warn about dense keys that never reached a parameter.
An adapter key the loader silently ignores is the failure mode this whole path
exists to fix, so it is said out loud rather than left to be inferred from a
model that merely generates worse.
"""
pending = (set(self._additive) | set(self._replacement)) - self._applied
if not pending:
logger.info("LoRA dense payload fully applied: %d parameters", len(self._applied))
return
for target in sorted(pending):
source = self._additive.get(target) or self._replacement[target]
logger.warning("LoRA key not loaded: %s (targets %s, absent from the model)", source[1], target)
logger.warning("LoRA dense payload: %d of %d parameters applied, %d keys unmatched", len(self._applied),
len(self._applied) + len(pending), len(pending))
def _read(self, entry: tuple[str, str]) -> torch.Tensor:
path, key = entry
with safe_open(path, framework="pt") as handle:
return handle.get_tensor(key)</code></pre></details>
fastvideo.models.loader.lora_patch.DenseLoRAPatch.from_adapter
from_adapter(lora_path: str | None, param_names_mapping: Callable[[str], tuple[str, Any, Any]] | None = None, *, lora_param_names_mapping: Callable[[str], tuple[str, Any, Any]] | None = None, strength: float = 1.0) -> DenseLoRAPatch | None
Build a patch from an adapter, or None when it carries no dense payload.
lora_param_names_mapping first translates adapter-specific official names
into the published checkpoint layout. param_names_mapping then resolves that
layout into the model's parameter names, matching the low-rank loader's order.
Source
@classmethod
def from_adapter(
cls,
lora_path: str | None,
param_names_mapping: Callable[[str], tuple[str, Any, Any]] | None = None,
*,
lora_param_names_mapping: Callable[[str], tuple[str, Any, Any]] | None = None,
strength: float = 1.0,
) -> DenseLoRAPatch | None:
"""Build a patch from an adapter, or ``None`` when it carries no dense payload.
``lora_param_names_mapping`` first translates adapter-specific official names
into the published checkpoint layout. ``param_names_mapping`` then resolves that
layout into the model's parameter names, matching the low-rank loader's order.
"""
if not lora_path:
return None
# Deferred: fastvideo.utils pulls in enough of the package that importing it at
# module scope would make this loader helper part of an import cycle.
from fastvideo.utils import maybe_download_lora
files = _adapter_files(maybe_download_lora(lora_path))
if not files:
logger.warning("LoRA path %s holds no safetensors file; no dense patch applied", lora_path)
return None
additive: dict[str, tuple[str, str]] = {}
replacement: dict[str, tuple[str, str]] = {}
for path in files:
with safe_open(path, framework="pt") as handle:
for key in handle.keys():
resolved = _resolve(key, lora_param_names_mapping, param_names_mapping)
if resolved is None:
continue
target, kind = resolved
table = additive if kind == "add" else replacement
if target in table:
raise ValueError(f"LoRA adapter {lora_path} maps two keys onto parameter {target}; "
f"the second is {key}")
table[target] = (path, key)
if not additive and not replacement:
return None
logger.info(
"LoRA adapter %s carries a dense payload: %d additive, %d replacement parameters",
lora_path, len(additive), len(replacement))
return cls(files, additive, replacement, strength)</code></pre></details>
fastvideo.models.loader.lora_patch.DenseLoRAPatch.apply_to
apply_to(param_name: str, tensor: torch.Tensor) -> torch.Tensor
Add this parameter's .diff/.diff_b delta, if the adapter has one.
The sum is taken in float32 whatever the operands are: both sides arrive in
bfloat16, whose 8-bit significand would quantize away a delta three orders of
magnitude below the base weight. The caller casts back to the target dtype.
Source
def apply_to(self, param_name: str, tensor: torch.Tensor) -> torch.Tensor:
"""Add this parameter's ``.diff``/``.diff_b`` delta, if the adapter has one.
The sum is taken in float32 whatever the operands are: both sides arrive in
bfloat16, whose 8-bit significand would quantize away a delta three orders of
magnitude below the base weight. The caller casts back to the target dtype.
"""
entry = self._additive.get(param_name)
if entry is None:
return tensor
delta = self._read(entry)
if delta.shape != tensor.shape:
raise ValueError(f"LoRA diff for {param_name} has shape {tuple(delta.shape)}, "
f"but the parameter is {tuple(tensor.shape)}")
self._applied.add(param_name)
# Safetensors returns the adapter delta on CPU, while inference can stream the
# base checkpoint directly onto its target GPU when CPU offload is disabled.
delta = delta.to(device=tensor.device, dtype=torch.float32)
return tensor.to(torch.float32) + delta * self._strength</code></pre></details>
fastvideo.models.loader.lora_patch.DenseLoRAPatch.provides
provides(param_name: str) -> bool
Whether the adapter carries this parameter whole, without reading it.
Source
def provides(self, param_name: str) -> bool:
"""Whether the adapter carries this parameter whole, without reading it."""
return param_name in self._replacement
fastvideo.models.loader.lora_patch.DenseLoRAPatch.replacement_parameters
replacement_parameters: frozenset[str]
Names of the parameters this adapter supplies outright.
Lets a caller decide what the adapter needs from the runtime -- an H3 adapter
carrying to_gate_compress only works under the VSA backend -- without
reading any tensor or guessing from the file name.
Source
def replacement_parameters(self) -> frozenset[str]:
"""Names of the parameters this adapter supplies outright.
Lets a caller decide what the adapter needs from the runtime -- an H3 adapter
carrying ``to_gate_compress`` only works under the VSA backend -- without
reading any tensor or guessing from the file name.
"""
return frozenset(self._replacement)</code></pre></details>
fastvideo.models.loader.lora_patch.DenseLoRAPatch.replacement_for
replacement_for(param_name: str) -> torch.Tensor | None
The adapter's whole-tensor value for a parameter, or None.
Source
def replacement_for(self, param_name: str) -> torch.Tensor | None:
"""The adapter's whole-tensor value for a parameter, or ``None``."""
entry = self._replacement.get(param_name)
if entry is None:
return None
self._applied.add(param_name)
# Replacement payloads address parameters absent from the base checkpoint. The
# loader initializes those parameters to zero, so scaling the supplied value is
# the same interpolation contract as ``base + strength * delta``.
return self._read(entry).to(torch.float32) * self._strength
fastvideo.models.loader.lora_patch.DenseLoRAPatch.report_unapplied
report_unapplied() -> None
Warn about dense keys that never reached a parameter.
An adapter key the loader silently ignores is the failure mode this whole path
exists to fix, so it is said out loud rather than left to be inferred from a
model that merely generates worse.
Source
def report_unapplied(self) -> None:
"""Warn about dense keys that never reached a parameter.
An adapter key the loader silently ignores is the failure mode this whole path
exists to fix, so it is said out loud rather than left to be inferred from a
model that merely generates worse.
"""
pending = (set(self._additive) | set(self._replacement)) - self._applied
if not pending:
logger.info("LoRA dense payload fully applied: %d parameters", len(self._applied))
return
for target in sorted(pending):
source = self._additive.get(target) or self._replacement[target]
logger.warning("LoRA key not loaded: %s (targets %s, absent from the model)", source[1], target)
logger.warning("LoRA dense payload: %d of %d parameters applied, %d keys unmatched", len(self._applied),
len(self._applied) + len(pending), len(pending))</code></pre></details>