lingbotworld2_t5
LingBot World 2 UMT5 encoder with the released checkpoint's module names.
fastvideo.models.encoders.lingbotworld2_t5.BaseEncoderOutput
BaseEncoderOutput(last_hidden_state: torch.FloatTensor | None = None, pooler_output: torch.FloatTensor | None = None, hidden_states: tuple[torch.FloatTensor, ...] | None = None, attentions: tuple[torch.FloatTensor, ...] | None = None, attention_mask: torch.Tensor | None = None)fastvideo.models.encoders.lingbotworld2_t5.LingBotWorld2UMT5Config
LingBotWorld2UMT5Config(arch_config: TextEncoderArchConfig = LingBotWorld2UMT5ArchConfig(), prefix: str = 'text_encoder', quant_config: QuantizationConfig | None = None, lora_config: Any | None = None, is_chat_model: bool = False, treat_empty_as_dot: bool = False, *, _resolved_attention_backend: AttentionBackendEnum | None = None, chat_template_enable_thinking: bool = False)fastvideo.models.encoders.lingbotworld2_t5.TextEncoder
TextEncoder(config: TextEncoderConfig)Base for native encoders with a model-specific forward output contract.
fastvideo.models.encoders.lingbotworld2_t5.TextEncoder.forward
forward(*args: Any, **kwargs: Any) -> TextEncoderOutputTfastvideo.models.encoders.lingbotworld2_t5.default_weight_loader
default_weight_loader(param: torch.Tensor, loaded_weight: torch.Tensor) -> NoneDefault weight loader.
fastvideo.models.encoders.lingbotworld2_t5.basic_clean
basic_clean(text: str) -> strApply LingBot World 2's ftfy/html cleanup before tokenization.
Source
def basic_clean(text: str) -> str:
"""Apply LingBot World 2's ftfy/html cleanup before tokenization."""
text = ftfy.fix_text(text)
text = html.unescape(html.unescape(text))
return text.strip()fastvideo.models.encoders.lingbotworld2_t5.whitespace_clean
whitespace_clean(text: str) -> strCollapse all whitespace runs to single spaces.
Source
def whitespace_clean(text: str) -> str:
"""Collapse all whitespace runs to single spaces."""
return " ".join(text.split())fastvideo.models.encoders.lingbotworld2_t5.canonicalize
canonicalize(text: str, keep_punctuation_exact_string: str | None = None) -> strNormalize prompts with LingBot World 2's optional punctuation handling.
Source
def canonicalize(text: str, keep_punctuation_exact_string: str | None = None) -> str:
"""Normalize prompts with LingBot World 2's optional punctuation handling."""
text = text.replace("_", " ")
if keep_punctuation_exact_string:
text = keep_punctuation_exact_string.join(
part.translate(str.maketrans("", "", string.punctuation))
for part in text.split(keep_punctuation_exact_string))
else:
text = text.translate(str.maketrans("", "", string.punctuation))
return " ".join(text.lower().split())fastvideo.models.encoders.lingbotworld2_t5.lingbotworld2_whitespace_preprocess
lingbotworld2_whitespace_preprocess(prompt: str) -> strMatch the LingBot World 2 source tokenizer's clean='whitespace' behavior.
Source
def lingbotworld2_whitespace_preprocess(prompt: str) -> str:
"""Match the LingBot World 2 source tokenizer's `clean='whitespace'` behavior."""
return whitespace_clean(basic_clean(prompt))fastvideo.models.encoders.lingbotworld2_t5.GELU
GELU()T5 gated GELU approximation used by the source checkpoint.
Source
class GELU(nn.Module):
"""T5 gated GELU approximation used by the source checkpoint."""
def forward(self, x: torch.Tensor) -> torch.Tensor:
"""Apply the tanh GELU approximation."""
return 0.5 * x * (1.0 + torch.tanh(math.sqrt(2.0 / math.pi) * (x + 0.044715 * torch.pow(x, 3.0))))</code></pre></details>
fastvideo.models.encoders.lingbotworld2_t5.GELU.forward
forward(x: torch.Tensor) -> torch.Tensor
Apply the tanh GELU approximation.
Source
def forward(self, x: torch.Tensor) -> torch.Tensor:
"""Apply the tanh GELU approximation."""
return 0.5 * x * (1.0 + torch.tanh(math.sqrt(2.0 / math.pi) * (x + 0.044715 * torch.pow(x, 3.0))))
fastvideo.models.encoders.lingbotworld2_t5.T5LayerNorm
T5LayerNorm(dim: int, eps: float = 1e-06)
T5 RMS-style layer norm with source-compatible parameter name.
Source
class T5LayerNorm(nn.Module):
"""T5 RMS-style layer norm with source-compatible parameter name."""
def __init__(self, dim: int, eps: float = 1e-6):
super().__init__()
self.dim = dim
self.eps = eps
self.weight = nn.Parameter(torch.ones(dim))
def forward(self, x: torch.Tensor) -> torch.Tensor:
"""Normalize in fp32 and apply the learned scale."""
x = x * torch.rsqrt(x.float().pow(2).mean(dim=-1, keepdim=True) + self.eps)
if self.weight.dtype in (torch.float16, torch.bfloat16):
x = x.type_as(self.weight)
return self.weight * x</code></pre></details>
fastvideo.models.encoders.lingbotworld2_t5.T5LayerNorm.forward
forward(x: torch.Tensor) -> torch.Tensor
Normalize in fp32 and apply the learned scale.
Source
def forward(self, x: torch.Tensor) -> torch.Tensor:
"""Normalize in fp32 and apply the learned scale."""
x = x * torch.rsqrt(x.float().pow(2).mean(dim=-1, keepdim=True) + self.eps)
if self.weight.dtype in (torch.float16, torch.bfloat16):
x = x.type_as(self.weight)
return self.weight * x
fastvideo.models.encoders.lingbotworld2_t5.T5Attention
T5Attention(dim: int, dim_attn: int, num_heads: int, dropout: float = 0.1)
LingBot World 2 source T5 attention block.
Source
class T5Attention(nn.Module):
"""LingBot World 2 source T5 attention block."""
def __init__(self, dim: int, dim_attn: int, num_heads: int, dropout: float = 0.1):
super().__init__()
assert dim_attn % num_heads == 0
self.dim = dim
self.dim_attn = dim_attn
self.num_heads = num_heads
self.head_dim = dim_attn // num_heads
self.q = nn.Linear(dim, dim_attn, bias=False)
self.k = nn.Linear(dim, dim_attn, bias=False)
self.v = nn.Linear(dim, dim_attn, bias=False)
self.o = nn.Linear(dim_attn, dim, bias=False)
self.dropout = nn.Dropout(dropout)
def forward(
self,
x: torch.Tensor,
context: torch.Tensor | None = None,
mask: torch.Tensor | None = None,
pos_bias: torch.Tensor | None = None,
) -> torch.Tensor:
"""Project QKV, add relative bias/mask, and return attended states."""
context = x if context is None else context
b, n, c = x.size(0), self.num_heads, self.head_dim
q = self.q(x).view(b, -1, n, c)
k = self.k(context).view(b, -1, n, c)
v = self.v(context).view(b, -1, n, c)
attn_bias = x.new_zeros(b, n, q.size(1), k.size(1))
if pos_bias is not None:
attn_bias += pos_bias
if mask is not None:
assert mask.ndim in (2, 3)
mask = mask.view(b, 1, 1, -1) if mask.ndim == 2 else mask.unsqueeze(1)
attn_bias.masked_fill_(mask == 0, torch.finfo(x.dtype).min)
attn = torch.einsum("binc,bjnc->bnij", q, k) + attn_bias
attn = F.softmax(attn.float(), dim=-1).type_as(attn)
x = torch.einsum("bnij,bjnc->binc", attn, v)
return self.dropout(self.o(x.reshape(b, -1, n * c)))</code></pre></details>
fastvideo.models.encoders.lingbotworld2_t5.T5Attention.forward
forward(x: torch.Tensor, context: torch.Tensor | None = None, mask: torch.Tensor | None = None, pos_bias: torch.Tensor | None = None) -> torch.Tensor
Project QKV, add relative bias/mask, and return attended states.
Source
def forward(
self,
x: torch.Tensor,
context: torch.Tensor | None = None,
mask: torch.Tensor | None = None,
pos_bias: torch.Tensor | None = None,
) -> torch.Tensor:
"""Project QKV, add relative bias/mask, and return attended states."""
context = x if context is None else context
b, n, c = x.size(0), self.num_heads, self.head_dim
q = self.q(x).view(b, -1, n, c)
k = self.k(context).view(b, -1, n, c)
v = self.v(context).view(b, -1, n, c)
attn_bias = x.new_zeros(b, n, q.size(1), k.size(1))
if pos_bias is not None:
attn_bias += pos_bias
if mask is not None:
assert mask.ndim in (2, 3)
mask = mask.view(b, 1, 1, -1) if mask.ndim == 2 else mask.unsqueeze(1)
attn_bias.masked_fill_(mask == 0, torch.finfo(x.dtype).min)
attn = torch.einsum("binc,bjnc->bnij", q, k) + attn_bias
attn = F.softmax(attn.float(), dim=-1).type_as(attn)
x = torch.einsum("bnij,bjnc->binc", attn, v)
return self.dropout(self.o(x.reshape(b, -1, n * c)))
fastvideo.models.encoders.lingbotworld2_t5.T5FeedForward
T5FeedForward(dim: int, dim_ffn: int, dropout: float = 0.1)
LingBot World 2 source T5 gated feed-forward block.
Source
class T5FeedForward(nn.Module):
"""LingBot World 2 source T5 gated feed-forward block."""
def __init__(self, dim: int, dim_ffn: int, dropout: float = 0.1):
super().__init__()
self.dim = dim
self.dim_ffn = dim_ffn
self.gate = nn.Sequential(nn.Linear(dim, dim_ffn, bias=False), GELU())
self.fc1 = nn.Linear(dim, dim_ffn, bias=False)
self.fc2 = nn.Linear(dim_ffn, dim, bias=False)
self.dropout = nn.Dropout(dropout)
def forward(self, x: torch.Tensor) -> torch.Tensor:
"""Apply the gated feed-forward projection."""
x = self.fc1(x) * self.gate(x)
x = self.dropout(x)
x = self.fc2(x)
return self.dropout(x)</code></pre></details>
fastvideo.models.encoders.lingbotworld2_t5.T5FeedForward.forward
forward(x: torch.Tensor) -> torch.Tensor
Apply the gated feed-forward projection.
Source
def forward(self, x: torch.Tensor) -> torch.Tensor:
"""Apply the gated feed-forward projection."""
x = self.fc1(x) * self.gate(x)
x = self.dropout(x)
x = self.fc2(x)
return self.dropout(x)
fastvideo.models.encoders.lingbotworld2_t5.T5RelativeEmbedding
T5RelativeEmbedding(num_buckets: int, num_heads: int, bidirectional: bool, max_dist: int = 128)
Per-block relative position embedding used by LingBot World 2 UMT5.
Source
class T5RelativeEmbedding(nn.Module):
"""Per-block relative position embedding used by LingBot World 2 UMT5."""
def __init__(self, num_buckets: int, num_heads: int, bidirectional: bool, max_dist: int = 128):
super().__init__()
self.num_buckets = num_buckets
self.num_heads = num_heads
self.bidirectional = bidirectional
self.max_dist = max_dist
self.embedding = nn.Embedding(num_buckets, num_heads)
def forward(self, lq: int, lk: int) -> torch.Tensor:
"""Build a relative-position bias tensor for attention logits."""
device = self.embedding.weight.device
rel_pos = torch.arange(lk, device=device).unsqueeze(0) - torch.arange(lq, device=device).unsqueeze(1)
rel_pos = self._relative_position_bucket(rel_pos)
rel_pos_embeds = self.embedding(rel_pos)
return rel_pos_embeds.permute(2, 0, 1).unsqueeze(0).contiguous()
def _relative_position_bucket(self, rel_pos: torch.Tensor) -> torch.Tensor:
"""Map token offsets to T5 relative-position buckets."""
if self.bidirectional:
num_buckets = self.num_buckets // 2
rel_buckets = (rel_pos > 0).long() * num_buckets
rel_pos = torch.abs(rel_pos)
else:
num_buckets = self.num_buckets
rel_buckets = 0
rel_pos = -torch.min(rel_pos, torch.zeros_like(rel_pos))
max_exact = num_buckets // 2
rel_pos_large = max_exact + (torch.log(rel_pos.float() / max_exact) / math.log(self.max_dist / max_exact) *
(num_buckets - max_exact)).long()
rel_pos_large = torch.min(rel_pos_large, torch.full_like(rel_pos_large, num_buckets - 1))
rel_buckets += torch.where(rel_pos < max_exact, rel_pos, rel_pos_large)
return rel_buckets</code></pre></details>
fastvideo.models.encoders.lingbotworld2_t5.T5RelativeEmbedding.forward
forward(lq: int, lk: int) -> torch.Tensor
Build a relative-position bias tensor for attention logits.
Source
def forward(self, lq: int, lk: int) -> torch.Tensor:
"""Build a relative-position bias tensor for attention logits."""
device = self.embedding.weight.device
rel_pos = torch.arange(lk, device=device).unsqueeze(0) - torch.arange(lq, device=device).unsqueeze(1)
rel_pos = self._relative_position_bucket(rel_pos)
rel_pos_embeds = self.embedding(rel_pos)
return rel_pos_embeds.permute(2, 0, 1).unsqueeze(0).contiguous()
fastvideo.models.encoders.lingbotworld2_t5.T5SelfAttention
T5SelfAttention(dim: int, dim_attn: int, dim_ffn: int, num_heads: int, num_buckets: int, shared_pos: bool = False, dropout: float = 0.1)
One source-compatible UMT5 encoder block.
Source
class T5SelfAttention(nn.Module):
"""One source-compatible UMT5 encoder block."""
def __init__(
self,
dim: int,
dim_attn: int,
dim_ffn: int,
num_heads: int,
num_buckets: int,
shared_pos: bool = False,
dropout: float = 0.1,
):
super().__init__()
self.norm1 = T5LayerNorm(dim)
self.attn = T5Attention(dim, dim_attn, num_heads, dropout)
self.norm2 = T5LayerNorm(dim)
self.ffn = T5FeedForward(dim, dim_ffn, dropout)
self.pos_embedding = None if shared_pos else T5RelativeEmbedding(num_buckets, num_heads, bidirectional=True)
def forward(self,
x: torch.Tensor,
mask: torch.Tensor | None = None,
pos_bias: torch.Tensor | None = None) -> torch.Tensor:
"""Run self-attention and feed-forward residual updates."""
e = pos_bias if self.pos_embedding is None else self.pos_embedding(x.size(1), x.size(1))
x = self._fp16_clamp(x + self.attn(self.norm1(x), mask=mask, pos_bias=e))
return self._fp16_clamp(x + self.ffn(self.norm2(x)))
@staticmethod
def _fp16_clamp(x: torch.Tensor) -> torch.Tensor:
if x.dtype == torch.float16 and torch.isinf(x).any():
clamp = torch.finfo(x.dtype).max - 1000
x = torch.clamp(x, min=-clamp, max=clamp)
return x</code></pre></details>
fastvideo.models.encoders.lingbotworld2_t5.T5SelfAttention.forward
forward(x: torch.Tensor, mask: torch.Tensor | None = None, pos_bias: torch.Tensor | None = None) -> torch.Tensor
Run self-attention and feed-forward residual updates.
Source
def forward(self,
x: torch.Tensor,
mask: torch.Tensor | None = None,
pos_bias: torch.Tensor | None = None) -> torch.Tensor:
"""Run self-attention and feed-forward residual updates."""
e = pos_bias if self.pos_embedding is None else self.pos_embedding(x.size(1), x.size(1))
x = self._fp16_clamp(x + self.attn(self.norm1(x), mask=mask, pos_bias=e))
return self._fp16_clamp(x + self.ffn(self.norm2(x)))
fastvideo.models.encoders.lingbotworld2_t5.LingBotWorld2T5EncoderModel
LingBotWorld2T5EncoderModel(config: LingBotWorld2UMT5Config, prefix: str = '')
FastVideo-native LingBot World 2 UMT5 encoder for the released .pth weights.
Source
class LingBotWorld2T5EncoderModel(TextEncoder):
"""FastVideo-native LingBot World 2 UMT5 encoder for the released `.pth` weights."""
fall_back_to_pt_during_load = True
allow_patterns_overrides = ["*.pt"]
def __init__(self, config: LingBotWorld2UMT5Config, prefix: str = ""):
super().__init__(config)
del prefix
arch = config.arch_config
self.token_embedding = nn.Embedding(arch.vocab_size, arch.dim)
self.dropout = nn.Dropout(arch.dropout)
self.blocks = nn.ModuleList([
T5SelfAttention(
arch.dim,
arch.dim_attn,
arch.dim_ffn,
arch.num_heads,
arch.num_buckets,
shared_pos=False,
dropout=arch.dropout,
) for _ in range(arch.num_layers)
])
self.norm = T5LayerNorm(arch.dim)
def forward(
self,
input_ids: torch.Tensor | None,
position_ids: torch.Tensor | None = None,
attention_mask: torch.Tensor | None = None,
inputs_embeds: torch.Tensor | None = None,
output_hidden_states: bool | None = None,
**kwargs,
) -> BaseEncoderOutput:
"""Encode token ids and return source-compatible hidden states."""
del position_ids, inputs_embeds, output_hidden_states, kwargs
assert input_ids is not None
x = self.dropout(self.token_embedding(input_ids))
for block in self.blocks:
x = block(x, attention_mask)
x = self.dropout(self.norm(x))
return BaseEncoderOutput(last_hidden_state=x, attention_mask=attention_mask)
def load_weights(self, weights: Iterable[tuple[str, torch.Tensor]]) -> set[str]:
"""Load source `.pth` weights whose names already match this module."""
params_dict = dict(self.named_parameters())
loaded_params: set[str] = set()
for name, loaded_weight in weights:
if name not in params_dict:
continue
param = params_dict[name]
weight_loader = getattr(param, "weight_loader", default_weight_loader)
weight_loader(param, loaded_weight)
loaded_params.add(name)
return loaded_params</code></pre></details>
fastvideo.models.encoders.lingbotworld2_t5.LingBotWorld2T5EncoderModel.forward
forward(input_ids: torch.Tensor | None, position_ids: torch.Tensor | None = None, attention_mask: torch.Tensor | None = None, inputs_embeds: torch.Tensor | None = None, output_hidden_states: bool | None = None, **kwargs) -> BaseEncoderOutput
Encode token ids and return source-compatible hidden states.
Source
def forward(
self,
input_ids: torch.Tensor | None,
position_ids: torch.Tensor | None = None,
attention_mask: torch.Tensor | None = None,
inputs_embeds: torch.Tensor | None = None,
output_hidden_states: bool | None = None,
**kwargs,
) -> BaseEncoderOutput:
"""Encode token ids and return source-compatible hidden states."""
del position_ids, inputs_embeds, output_hidden_states, kwargs
assert input_ids is not None
x = self.dropout(self.token_embedding(input_ids))
for block in self.blocks:
x = block(x, attention_mask)
x = self.dropout(self.norm(x))
return BaseEncoderOutput(last_hidden_state=x, attention_mask=attention_mask)
fastvideo.models.encoders.lingbotworld2_t5.LingBotWorld2T5EncoderModel.load_weights
load_weights(weights: Iterable[tuple[str, torch.Tensor]]) -> set[str]
Load source .pth weights whose names already match this module.
Source
def load_weights(self, weights: Iterable[tuple[str, torch.Tensor]]) -> set[str]:
"""Load source `.pth` weights whose names already match this module."""
params_dict = dict(self.named_parameters())
loaded_params: set[str] = set()
for name, loaded_weight in weights:
if name not in params_dict:
continue
param = params_dict[name]
weight_loader = getattr(param, "weight_loader", default_weight_loader)
weight_loader(param, loaded_weight)
loaded_params.add(name)
return loaded_params