--- title: 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) -> TextEncoderOutputT

fastvideo.models.encoders.lingbotworld2_t5.default_weight_loader

default_weight_loader(param: torch.Tensor, loaded_weight: torch.Tensor) -> None

Default weight loader.

fastvideo.models.encoders.lingbotworld2_t5.basic_clean

basic_clean(text: str) -> str

Apply 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) -> str

Collapse 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) -> str

Normalize 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) -> str

Match 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))))

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

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)))

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)

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

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

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

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