cleanup and local test

This commit is contained in:
H1yori233
2026-07-07 15:00:49 -07:00
committed by SolitaryThinker
parent 2d7e262b13
commit a0749e98fd
28 changed files with 2085 additions and 955 deletions
+4 -12
View File
@@ -1,13 +1,4 @@
# SPDX-License-Identifier: Apache-2.0
"""
GLM-Image inference example using FastVideo.
This example demonstrates how to generate images using the GLM-Image model
through FastVideo's VideoGenerator API.
Usage:
python examples/inference/basic/basic_glm_image.py
"""
import os
@@ -41,11 +32,12 @@ def main() -> None:
height=1024,
width=1024,
num_inference_steps=50,
guidance_scale=7.5,
guidance_scale=1.5,
)
if result and len(result) > 0:
img = Image.fromarray(result[0])
frames = result.get("frames") if isinstance(result, dict) else None
if frames:
img = Image.fromarray(frames[0])
img.save(os.path.join(OUTPUT_PATH, "landscape.png"))
print(f"Saved image to {OUTPUT_PATH}/landscape.png")
@@ -0,0 +1,47 @@
# SPDX-License-Identifier: Apache-2.0
import os
from PIL import Image
from fastvideo import VideoGenerator
OUTPUT_PATH = "image_output"
CONDITION_IMAGE = "assets/images/couple.jpg"
PROMPT = "Change the background to a snowy mountain landscape at golden hour."
def main() -> None:
generator = VideoGenerator.from_pretrained(
"zai-org/GLM-Image",
num_gpus=1,
trust_remote_code=True,
)
os.makedirs(OUTPUT_PATH, exist_ok=True)
condition = Image.open(CONDITION_IMAGE).convert("RGB")
result = generator.generate_video(
prompt=PROMPT,
pil_image=condition,
output_path=OUTPUT_PATH,
save_video=False,
return_frames=True,
height=1024,
width=1024,
num_inference_steps=50,
guidance_scale=1.5,
)
frames = result.get("frames") if isinstance(result, dict) else None
if frames:
img = Image.fromarray(frames[0])
img.save(os.path.join(OUTPUT_PATH, "edited.png"))
print(f"Saved image to {OUTPUT_PATH}/edited.png")
generator.shutdown()
if __name__ == "__main__":
main()
+5 -50
View File
@@ -1,9 +1,4 @@
# SPDX-License-Identifier: Apache-2.0
"""GLM-Image DiT configuration.
GLM-Image uses a 7B diffusion transformer (DiT) decoder that expands tokens
from the autoregressive vision-language encoder into high-resolution images.
"""
from dataclasses import dataclass, field
from fastvideo.configs.models.dits.base import DiTArchConfig, DiTConfig
@@ -15,12 +10,9 @@ def is_blocks(n: str, m) -> bool:
@dataclass
class GlmImageDiTArchConfig(DiTArchConfig):
"""Architecture config for GlmImageTransformer2DModel."""
_fsdp_shard_conditions: list = field(default_factory=lambda: [is_blocks])
# GLM-Image DiT settings (7B model)
# hidden_size = num_attention_heads * attention_head_dim = 64 * 64 = 4096
hidden_size: int = 4096
num_attention_heads: int = 32
attention_head_dim: int = 128
@@ -28,66 +20,31 @@ class GlmImageDiTArchConfig(DiTArchConfig):
out_channels: int = 16
num_layers: int = 30
# Text and condition dims
text_embed_dim: int = 1472
time_embed_dim: int = 512
condition_dim: int = 256
# VQ settings for AR tokens
prior_vq_quantizer_codebook_size: int = 16384
# Patch embedding
patch_size: int = 2
# Positional embedding max resolution
max_height: int = 2048
max_width: int = 2048
# QK normalization
qk_norm: str = "layer_norm"
eps: float = 1e-5
# LoRA exclusions
exclude_lora_layers: list[str] = field(
default_factory=lambda:
["image_projector", "glyph_projector", "prior_token_embedding"])
default_factory=lambda: ["image_projector", "glyph_projector", "prior_token_embedding"])
# Param name mappings for weight loading (HF -> custom)
param_names_mapping: dict = field(
default_factory=lambda: {
# Projectors (mapped from FeedForward net.0.proj -> fc_in, net.2 -> fc_out)
r"^image_projector\.net\.0\.proj\.(.*)$":
r"image_projector.fc_in.\1",
r"^image_projector\.net\.2\.(.*)$": r"image_projector.fc_out.\1",
r"^glyph_projector\.net\.0\.proj\.(.*)$":
r"glyph_projector.fc_in.\1",
r"^glyph_projector\.net\.0\.proj\.(.*)$": r"glyph_projector.fc_in.\1",
r"^glyph_projector\.net\.2\.(.*)$": r"glyph_projector.fc_out.\1",
r"^prior_projector\.net\.0\.proj\.(.*)$":
r"prior_projector.fc_in.\1",
r"^prior_projector\.net\.0\.proj\.(.*)$": r"prior_projector.fc_in.\1",
r"^prior_projector\.net\.2\.(.*)$": r"prior_projector.fc_out.\1",
r"^prior_token_embedding\.(.*)$": r"prior_token_embedding.\1",
# Transformer blocks
r"^transformer_blocks\.(\d+)\.norm1\.(.*)$":
r"transformer_blocks.\1.norm1.\2",
r"^transformer_blocks\.(\d+)\.attn1\.to_q\.(.*)$":
r"transformer_blocks.\1.attn1.to_q.\2",
r"^transformer_blocks\.(\d+)\.attn1\.to_k\.(.*)$":
r"transformer_blocks.\1.attn1.to_k.\2",
r"^transformer_blocks\.(\d+)\.attn1\.to_v\.(.*)$":
r"transformer_blocks.\1.attn1.to_v.\2",
r"^transformer_blocks\.(\d+)\.attn1\.to_out\.0\.(.*)$":
r"transformer_blocks.\1.attn1.to_out.0.\2",
# FeedForward in blocks (net.0.proj -> fc_in, net.2 -> fc_out)
r"^transformer_blocks\.(\d+)\.ff\.net\.0\.proj\.(.*)$":
r"transformer_blocks.\1.ff.fc_in.\2",
r"^transformer_blocks\.(\d+)\.ff\.net\.2\.(.*)$":
r"transformer_blocks.\1.ff.fc_out.\2",
# Output
r"^norm_out\.(.*)$": r"norm_out.\1",
r"^proj_out\.(.*)$": r"proj_out.\1",
r"^transformer_blocks\.(\d+)\.ff\.net\.0\.proj\.(.*)$": r"transformer_blocks.\1.ff.fc_in.\2",
r"^transformer_blocks\.(\d+)\.ff\.net\.2\.(.*)$": r"transformer_blocks.\1.ff.fc_out.\2",
})
reverse_param_names_mapping: dict = field(default_factory=dict)
@@ -100,7 +57,5 @@ class GlmImageDiTArchConfig(DiTArchConfig):
@dataclass
class GlmImageDiTConfig(DiTConfig):
"""Configuration for GLM-Image DiT model."""
arch_config: DiTArchConfig = field(default_factory=GlmImageDiTArchConfig)
prefix: str = "GlmImage"
+66 -25
View File
@@ -1,54 +1,95 @@
# SPDX-License-Identifier: Apache-2.0
"""GLM-Image VAE configuration.
from __future__ import annotations
GLM-Image uses an AutoencoderKL for encoding/decoding images to/from latent space.
This is an image-only VAE (no temporal compression).
"""
from dataclasses import dataclass, field
from fastvideo.configs.models.vaes.base import VAEArchConfig, VAEConfig
from fastvideo.configs.models.vaes.autoencoder_kl import (AutoencoderKLArchConfig, AutoencoderKLVAEConfig)
_GLM_IMAGE_LATENTS_MEAN: tuple[float, ...] = (
-0.2080078125,
1.875,
-0.470703125,
-1.265625,
-1.421875,
0.77734375,
-0.3671875,
-0.9453125,
0.318359375,
0.7734375,
-0.1884765625,
-0.022216796875,
-0.220703125,
-1.59375,
-0.81640625,
-0.255859375,
)
_GLM_IMAGE_LATENTS_STD: tuple[float, ...] = (
3.0625,
2.203125,
2.265625,
4.84375,
2.5,
3.9375,
2.203125,
3.03125,
2.1875,
2.046875,
2.71875,
2.390625,
2.390625,
2.453125,
2.25,
2.15625,
)
@dataclass
class GlmImageVAEArchConfig(VAEArchConfig):
"""Architecture config for GLM-Image VAE (AutoencoderKL)."""
# Standard KL-VAE parameters
class GlmImageVAEArchConfig(AutoencoderKLArchConfig):
act_fn: str = "silu"
block_out_channels: tuple[int, ...] = (128, 512, 1024, 1024)
down_block_types: tuple[str, ...] = (
"DownEncoderBlock2D",
"DownEncoderBlock2D",
"DownEncoderBlock2D",
"DownEncoderBlock2D",
)
up_block_types: tuple[str, ...] = (
"UpDecoderBlock2D",
"UpDecoderBlock2D",
"UpDecoderBlock2D",
"UpDecoderBlock2D",
)
force_upcast: bool = True
in_channels: int = 3
out_channels: int = 3
latent_channels: int = 16
# Encoder/decoder architecture
block_out_channels: tuple[int, ...] = (128, 256, 512, 512)
layers_per_block: int = 2
latents_mean: tuple[float, ...] = _GLM_IMAGE_LATENTS_MEAN
latents_std: tuple[float, ...] = _GLM_IMAGE_LATENTS_STD
layers_per_block: int = 3
mid_block_add_attention: bool = False
norm_num_groups: int = 32
# Scaling factor for latents (standard for SD-style VAEs)
sample_size: int = 1024
scaling_factor: float = 0.18215
shift_factor: float | None = None
use_quant_conv: bool = False
use_post_quant_conv: bool = False
# Image-only VAE: no temporal compression
temporal_compression_ratio: int = 1
spatial_compression_ratio: int = 8
@dataclass
class GlmImageVAEConfig(VAEConfig):
"""Configuration for GLM-Image VAE."""
class GlmImageVAEConfig(AutoencoderKLVAEConfig):
arch_config: GlmImageVAEArchConfig = field(default_factory=GlmImageVAEArchConfig)
arch_config: GlmImageVAEArchConfig = field(
default_factory=GlmImageVAEArchConfig)
# Tiling settings for high-resolution images
use_tiling: bool = True
use_temporal_tiling: bool = False # Image model, no temporal dimension
use_temporal_tiling: bool = False
use_parallel_tiling: bool = False
# Tile dimensions for memory efficiency
tile_sample_min_height: int = 512
tile_sample_min_width: int = 512
tile_sample_stride_height: int = 384
tile_sample_stride_width: int = 384
# For image models, we need both encoder and decoder
load_encoder: bool = True
load_decoder: bool = True
+7 -83
View File
@@ -1,14 +1,4 @@
# SPDX-License-Identifier: Apache-2.0
"""GLM-Image pipeline configuration.
GLM-Image is a hybrid model combining:
- 9B autoregressive (AR) vision-language encoder for semantic/layout token generation
- 7B diffusion transformer (DiT) decoder for high-fidelity image generation
The pipeline supports:
- Text-to-image (T2I) generation
- Image-to-image (I2I) editing/style transfer
"""
from collections.abc import Callable
from dataclasses import dataclass, field
@@ -22,101 +12,35 @@ from fastvideo.configs.pipelines.base import PipelineConfig
def glm_image_t5_postprocess(outputs: BaseEncoderOutput) -> torch.Tensor:
"""Postprocess T5 encoder outputs for GLM-Image.
GLM-Image uses T5 as a glyph/text encoder for text rendering capabilities.
"""
mask: torch.Tensor = outputs.attention_mask
hidden_state: torch.Tensor = outputs.last_hidden_state
seq_lens = mask.gt(0).sum(dim=1).long()
assert torch.isnan(hidden_state).sum() == 0, "T5 hidden states contain NaN"
# Pack embeddings with proper padding
prompt_embeds = [u[:v] for u, v in zip(hidden_state, seq_lens, strict=True)]
max_len = 512 # GLM-Image uses 512 max sequence length
prompt_embeds_tensor: torch.Tensor = torch.stack([
torch.cat([u, u.new_zeros(max_len - u.size(0), u.size(1))])
for u in prompt_embeds
],
dim=0)
max_len = 512
prompt_embeds_tensor: torch.Tensor = torch.stack(
[torch.cat([u, u.new_zeros(max_len - u.size(0), u.size(1))]) for u in prompt_embeds], dim=0)
return prompt_embeds_tensor
@dataclass
class GlmImageConfig(PipelineConfig):
"""Configuration for GLM-Image T2I/I2I pipeline.
This configuration sets up the complete GLM-Image pipeline including:
- T5 text encoder (glyph encoder for text rendering)
- Vision-language encoder (AR model for semantic tokens)
- DiT diffusion decoder
- VAE for image encoding/decoding
"""
# Image generation flag (not video)
is_video_pipeline: bool = False
# DiT configuration
dit_config: DiTConfig = field(default_factory=GlmImageDiTConfig)
dit_precision: str = "bf16"
# VAE configuration
vae_config: VAEConfig = field(default_factory=GlmImageVAEConfig)
vae_precision: str = "fp32"
vae_tiling: bool = True
vae_sp: bool = False # No spatial parallelism for image VAE
vae_sp: bool = False
# Text encoder (T5 for glyph/text rendering)
text_encoder_configs: tuple[EncoderConfig, ...] = field(
default_factory=lambda: (T5Config(), ))
text_encoder_precisions: tuple[str, ...] = field(
default_factory=lambda: ("fp32", ))
text_encoder_configs: tuple[EncoderConfig, ...] = field(default_factory=lambda: (T5Config(), ))
text_encoder_precisions: tuple[str, ...] = field(default_factory=lambda: ("fp32", ))
postprocess_text_funcs: tuple[Callable[[BaseEncoderOutput], torch.Tensor],
...] = field(default_factory=lambda:
(glm_image_t5_postprocess, ))
...] = field(default_factory=lambda: (glm_image_t5_postprocess, ))
# Flow matching scheduler settings
flow_shift: float | None = 1.0
embedded_cfg_scale: float = 7.5
# AR model sampling parameters
ar_temperature: float = 0.95
ar_top_p: float = 0.75
ar_do_sample: bool = True
ar_max_new_tokens: int = 1024
# Image resolution constraints (must be divisible by 32)
default_height: int = 1024
default_width: int = 1024
resolution_multiple: int = 32
# Post-decoding hook for custom processing
post_decoding: Callable | None = None
def __post_init__(self):
# GLM-Image needs both encoder and decoder
self.vae_config.load_encoder = True
self.vae_config.load_decoder = True
def validate_resolution(self, height: int, width: int) -> tuple[int, int]:
"""Ensure resolution is divisible by 32 as required by GLM-Image."""
h = (height // self.resolution_multiple) * self.resolution_multiple
w = (width // self.resolution_multiple) * self.resolution_multiple
return h, w
@dataclass
class GlmImageI2IConfig(GlmImageConfig):
"""Configuration for GLM-Image image-to-image pipeline.
Inherits from GlmImageConfig with settings optimized for I2I tasks
like style transfer, editing, and inpainting.
"""
# Lower CFG for I2I to preserve input structure
embedded_cfg_scale: float = 5.0
# Strength for I2I (how much of the original to preserve)
strength: float = 0.8
+36 -77
View File
@@ -1,11 +1,4 @@
# SPDX-License-Identifier: Apache-2.0
# Adapted from SGLang: https://github.com/sgl-project/sglang/blob/main/python/sglang/multimodal_gen/runtime/models/dits/glm_image.py
"""GLM-Image Transformer 2D Model.
This module implements the diffusion transformer (DiT) decoder for GLM-Image.
The model takes tokens from the autoregressive vision-language encoder and
denoises them into high-fidelity images.
"""
from typing import Any, Dict, List, Optional, Tuple, Union
import torch
@@ -19,31 +12,25 @@ from fastvideo.layers.layernorm import ScaleResidualLayerNormScaleShift
from fastvideo.layers.linear import ReplicatedLinear
from fastvideo.layers.rotary_embedding import _apply_rotary_emb
from fastvideo.layers.visual_embedding import Timesteps
from fastvideo.logger import init_logger
from fastvideo.models.dits.base import BaseDiT
from fastvideo.platforms import AttentionBackendEnum
logger = init_logger(__name__)
# =============================================================================
# KV Cache (for future use with caching)
# =============================================================================
class GlmImageLayerKVCache:
"""KV cache for GlmImage model per layer."""
def __init__(self):
self.k_cache = None
self.v_cache = None
self.mode: Optional[str] = None # "write", "read", "skip"
self.mode: Optional[str] = None
def store(self, k: torch.Tensor, v: torch.Tensor):
# Append along seq (dim=1).
if self.k_cache is None:
self.k_cache = k
self.v_cache = v
else:
self.k_cache = torch.cat([self.k_cache, k], dim=2)
self.v_cache = torch.cat([self.v_cache, v], dim=2)
self.k_cache = torch.cat([self.k_cache, k], dim=1)
self.v_cache = torch.cat([self.v_cache, v], dim=1)
def get(self):
return self.k_cache, self.v_cache
@@ -55,7 +42,6 @@ class GlmImageLayerKVCache:
class GlmImageKVCache:
"""Container for all layers' KV caches."""
def __init__(self, num_layers: int):
self.num_layers = num_layers
@@ -81,10 +67,6 @@ class GlmImageKVCache:
# Timestep and Text Projection
# =============================================================================
class GlmImageTimestepEmbedding(nn.Module):
"""
Replacement for diffusers TimestepEmbedding using ReplicatedLinear.
Structure: linear_1 -> act(silu) -> linear_2
"""
def __init__(
self,
@@ -113,10 +95,6 @@ class GlmImageTimestepEmbedding(nn.Module):
class GlmImageTextProjection(nn.Module):
"""
Replacement for diffusers PixArtAlphaTextProjection using ReplicatedLinear.
Structure: linear_1 -> act_1 -> linear_2
"""
def __init__(
self,
@@ -145,7 +123,6 @@ class GlmImageTextProjection(nn.Module):
class GlmImageCombinedTimestepSizeEmbeddings(nn.Module):
"""Combined timestep and size embeddings using Timesteps class."""
def __init__(
self,
@@ -185,15 +162,14 @@ class GlmImageCombinedTimestepSizeEmbeddings(nn.Module):
target_size.size(0), -1
)
# (B, 2 * condition_dim)
condition_proj = torch.cat([crop_coords_proj, target_size_proj], dim=1)
timesteps_emb = self.timestep_embedder(
timesteps_proj.to(dtype=hidden_dtype)
) # (B, embedding_dim)
)
condition_emb = self.condition_embedder(
condition_proj.to(dtype=hidden_dtype)
) # (B, embedding_dim)
)
conditioning = timesteps_emb + condition_emb
return conditioning
@@ -203,7 +179,6 @@ class GlmImageCombinedTimestepSizeEmbeddings(nn.Module):
# Image Projector
# =============================================================================
class GlmImageImageProjector(nn.Module):
"""Projects latent patches to hidden dimension."""
def __init__(
self,
@@ -240,7 +215,6 @@ class GlmImageImageProjector(nn.Module):
# AdaLayerNorm
# =============================================================================
class GlmImageAdaLayerNormZero(nn.Module):
"""Ada Layer Norm with 12-way modulation using ReplicatedLinear."""
def __init__(self, embedding_dim: int, dim: int) -> None:
super().__init__()
@@ -302,7 +276,6 @@ class GlmImageAdaLayerNormZero(nn.Module):
# Attention
# =============================================================================
class GlmImageAttention(nn.Module):
"""GLM-Image attention using LocalAttention backend with ReplicatedLinear."""
def __init__(
self,
@@ -400,10 +373,11 @@ class GlmImageAttention(nn.Module):
if kv_cache.mode == "write":
kv_cache.store(key, value)
elif kv_cache.mode == "read":
# Prepend cached condition k/v along seq (dim=1).
k_cache, v_cache = kv_cache.get()
key = torch.cat([k_cache, key], dim=2) if k_cache is not None else key
key = torch.cat([k_cache, key], dim=1) if k_cache is not None else key
value = (
torch.cat([v_cache, value], dim=2) if v_cache is not None else value
torch.cat([v_cache, value], dim=1) if v_cache is not None else value
)
elif kv_cache.mode == "skip":
pass
@@ -425,7 +399,6 @@ class GlmImageAttention(nn.Module):
# Transformer Block
# =============================================================================
class GlmImageTransformerBlock(nn.Module):
"""GLM-Image Transformer block with fused residual+norm operations."""
def __init__(
self,
@@ -504,7 +477,6 @@ class GlmImageTransformerBlock(nn.Module):
)
# 3. Feedforward (fused residual + norm + scale/shift)
# Fix: hidden_states is the "residual" (base stream), attn_hidden_states is the "x" (modulated by gate)
norm_hidden_states, hidden_states = self.norm2(
hidden_states,
attn_hidden_states,
@@ -534,26 +506,33 @@ class GlmImageTransformerBlock(nn.Module):
# Rotary Positional Embedding
# =============================================================================
class GlmImageRotaryPosEmbed(nn.Module):
"""2D rotary positional embeddings for image patches."""
def __init__(self, dim: int, patch_size: int, theta: float = 10000.0) -> None:
super().__init__()
self.dim = dim
self.patch_size = patch_size
self.theta = theta
self._cache_key: tuple | None = None
self._cache_value: tuple[torch.Tensor, torch.Tensor] | None = None
def forward(self, hidden_states: torch.Tensor) -> Tuple[torch.Tensor, torch.Tensor]:
batch_size, num_channels, height, width = hidden_states.shape
height, width = height // self.patch_size, width // self.patch_size
_, _, raw_h, raw_w = hidden_states.shape
height = raw_h // self.patch_size
width = raw_w // self.patch_size
device = hidden_states.device
cache_key = (height, width, device.type,
device.index if device.index is not None else -1)
if self._cache_key == cache_key and self._cache_value is not None:
return self._cache_value
dim_h, dim_w = self.dim // 2, self.dim // 2
h_inv_freq = 1.0 / (
self.theta
** (
torch.arange(0, dim_h, 2, dtype=torch.float32, device=device)[
: (dim_h // 2)
].float()
]
/ dim_h
)
)
@@ -562,36 +541,26 @@ class GlmImageRotaryPosEmbed(nn.Module):
** (
torch.arange(0, dim_w, 2, dtype=torch.float32, device=device)[
: (dim_w // 2)
].float()
]
/ dim_w
)
)
h_seq = torch.arange(height, device=device)
w_seq = torch.arange(width, device=device)
freqs_h = torch.outer(h_seq, h_inv_freq)
freqs_w = torch.outer(w_seq, w_inv_freq)
freqs_h = torch.outer(h_seq, h_inv_freq).unsqueeze(1).expand(height, width, -1)
freqs_w = torch.outer(w_seq, w_inv_freq).unsqueeze(0).expand(height, width, -1)
# Create position matrices for height and width
freqs_h = freqs_h.unsqueeze(1)
freqs_w = freqs_w.unsqueeze(0)
# Broadcast freqs_h and freqs_w to [height, width, dim//4]
freqs_h = freqs_h.expand(height, width, -1)
freqs_w = freqs_w.expand(height, width, -1)
# Concatenate along last dimension to get [height, width, dim//2]
freqs = torch.cat([freqs_h, freqs_w], dim=-1)
freqs = freqs.reshape(height * width, -1) # [height * width, dim//2]
return (freqs.cos(), freqs.sin())
freqs = torch.cat([freqs_h, freqs_w], dim=-1).reshape(height * width, -1)
result = (freqs.cos(), freqs.sin())
self._cache_key = cache_key
self._cache_value = result
return result
# =============================================================================
# Final AdaLayerNorm
# =============================================================================
class GlmImageAdaLayerNormContinuous(nn.Module):
"""
GlmImage-only final AdaLN: LN(x) -> Linear(cond) -> chunk -> affine.
Matches Megatron: **no activation** before the Linear on conditioning embedding.
"""
def __init__(
self,
@@ -616,7 +585,6 @@ class GlmImageAdaLayerNormContinuous(nn.Module):
def forward(
self, x: torch.Tensor, conditioning_embedding: torch.Tensor
) -> torch.Tensor:
# *** NO SiLU here ***
emb = self.linear(conditioning_embedding.to(x.dtype))
scale, shift = torch.chunk(emb, 2, dim=1)
x = self.norm(x) * (1 + scale)[:, None, :] + shift[:, None, :]
@@ -627,12 +595,7 @@ class GlmImageAdaLayerNormContinuous(nn.Module):
# Main Model
# =============================================================================
class GlmImageTransformer2DModel(BaseDiT):
"""
GLM-Image Transformer 2D Model.
Aligned with SGLang implementation for numerical consistency.
"""
_fsdp_shard_conditions = GlmImageDiTConfig().arch_config._fsdp_shard_conditions
_compile_conditions = GlmImageDiTConfig().arch_config._fsdp_shard_conditions
_supported_attention_backends = (
@@ -651,7 +614,6 @@ class GlmImageTransformer2DModel(BaseDiT):
):
super().__init__(config=config, hf_config=hf_config)
self.config_data = config # Store config
arch_config = config.arch_config
self.in_channels = arch_config.in_channels
@@ -668,7 +630,6 @@ class GlmImageTransformer2DModel(BaseDiT):
pooled_projection_dim = 2 * 2 * arch_config.condition_dim
inner_dim = arch_config.num_attention_heads * arch_config.attention_head_dim
# Required by BaseDiT
self.hidden_size = inner_dim
self.num_channels_latents = arch_config.out_channels
@@ -685,7 +646,7 @@ class GlmImageTransformer2DModel(BaseDiT):
input_dim=arch_config.text_embed_dim,
mlp_hidden_dim=inner_dim,
output_dim=inner_dim,
act_type="gelu_pytorch_tanh",
act_type="gelu",
)
self.prior_token_embedding = nn.Embedding(
arch_config.prior_vq_quantizer_codebook_size, inner_dim
@@ -737,9 +698,9 @@ class GlmImageTransformer2DModel(BaseDiT):
self,
hidden_states: torch.Tensor,
encoder_hidden_states: torch.Tensor,
timestep: torch.LongTensor,
prior_token_id: torch.Tensor,
prior_token_drop: torch.Tensor,
timestep: torch.LongTensor,
target_size: torch.Tensor,
crop_coords: torch.Tensor,
attention_kwargs: Optional[Dict[str, Any]] = None,
@@ -759,9 +720,6 @@ class GlmImageTransformer2DModel(BaseDiT):
batch_size, num_channels, height, width = hidden_states.shape
# SGLang does timestep -= 1.0
timestep = timestep - 1.0
if isinstance(encoder_hidden_states, list):
encoder_hidden_states = encoder_hidden_states[0]
@@ -778,10 +736,11 @@ class GlmImageTransformer2DModel(BaseDiT):
hidden_states = self.image_projector(hidden_states)
encoder_hidden_states = self.glyph_projector(encoder_hidden_states)
prior_embedding = self.prior_token_embedding(prior_token_id)
# Using boolean indexing for sample-level dropping
if prior_token_drop.any():
# Create a mask for the batch dimension
prior_embedding[prior_token_drop] *= 0.0
# Zero dropped priors by multiply: boolean indexing + .any() syncs each step.
keep = (~prior_token_drop).to(prior_embedding.dtype)
while keep.dim() < prior_embedding.dim():
keep = keep.unsqueeze(-1)
prior_embedding = prior_embedding * keep
prior_hidden_states = self.prior_projector(prior_embedding)
hidden_states = hidden_states + prior_hidden_states
@@ -0,0 +1,64 @@
# SPDX-License-Identifier: Apache-2.0
"""Sole HF-import boundary for GLM-Image's AR encoder (lazy-wrapper exception E001)."""
from __future__ import annotations
from typing import Any
import torch
import torch.nn as nn
from fastvideo.logger import init_logger
logger = init_logger(__name__)
class GlmImageARLoader(nn.Module):
def __init__(self, model_path: str, processor_path: str | None = None,
*, torch_dtype: torch.dtype = torch.bfloat16,
trust_remote_code: bool = True) -> None:
super().__init__()
from transformers import (AutoProcessor,
GlmImageForConditionalGeneration)
logger.info("Loading GLM-Image AR encoder from %s", model_path)
self._model = GlmImageForConditionalGeneration.from_pretrained(
model_path,
torch_dtype=torch_dtype,
trust_remote_code=trust_remote_code,
)
if processor_path is not None:
logger.info("Loading GLM-Image processor from %s", processor_path)
self.processor = AutoProcessor.from_pretrained(
processor_path, trust_remote_code=trust_remote_code)
else:
self.processor = None
@torch.no_grad()
def generate(self, *args: Any, **kwargs: Any) -> torch.Tensor:
return self._model.generate(*args, **kwargs)
@torch.no_grad()
def get_image_features(self, pixel_values: torch.Tensor,
image_grid_thw: torch.Tensor) -> Any:
return self._model.get_image_features(pixel_values, image_grid_thw)
@torch.no_grad()
def get_image_tokens(self, image_embeds: torch.Tensor,
image_grid_thw: torch.Tensor) -> torch.Tensor:
return self._model.get_image_tokens(image_embeds, image_grid_thw)
@property
def config(self): # type: ignore[no-untyped-def]
return self._model.config
@property
def generation_config(self): # type: ignore[no-untyped-def]
return self._model.generation_config
def to(self, *args, **kwargs): # type: ignore[override]
self._model = self._model.to(*args, **kwargs)
return super().to(*args, **kwargs)
def eval(self): # type: ignore[override]
self._model = self._model.eval()
return super().eval()
-1
View File
@@ -92,7 +92,6 @@ _IMAGE_ENCODER_MODELS: dict[str, tuple] = {
}
_VAE_MODELS = {
"AutoencoderKL": ("vaes", "autoencoderkl", "AutoencoderKL"), # Generic 2D VAE
"AutoencoderKLHunyuanVideo":
("vaes", "hunyuanvae", "AutoencoderKLHunyuanVideo"),
"AutoencoderKLCausal3D": ("vaes", "gamecraftvae", "GameCraftVAE"),
@@ -1,18 +1,16 @@
# SPDX-License-Identifier: Apache-2.0
"""
GLM-Image diffusion pipeline implementation.
"""
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.logger import init_logger
from fastvideo.models.schedulers.scheduling_flow_match_euler_discrete import (
FlowMatchEulerDiscreteScheduler, )
from fastvideo.pipelines import ComposedPipelineBase, LoRAPipeline
from fastvideo.pipelines.basic.glm_image.stages import (
GlmImageBeforeDenoisingStage,
GlmImageConditionEncodingStage,
GlmImageDecodingStage,
GlmImageDenoisingStage,
)
from fastvideo.pipelines.stages import InputValidationStage
from fastvideo.pipelines.stages.glm_image_before_denoising import (
GlmImageBeforeDenoisingStage, )
from fastvideo.pipelines.stages.glm_image_denoising import GlmImageDenoisingStage
from fastvideo.pipelines.stages.glm_image_decoding import GlmImageDecodingStage
logger = init_logger(__name__)
@@ -26,21 +24,15 @@ class GlmImagePipeline(LoRAPipeline, ComposedPipelineBase):
"vae",
"transformer",
"scheduler",
"vision_language_encoder", # AR model for prior tokens - critical for quality
"processor", # Processor for AR model
"vision_language_encoder",
"processor",
]
_optional_config_modules: list[str] = []
def initialize_pipeline(self, fastvideo_args: FastVideoArgs) -> None:
# Initial shift, will be updated dynamically in BeforeDenoisingStage
self.modules["scheduler"] = FlowMatchEulerDiscreteScheduler(shift=1.0)
# Validate configuration if provided
if fastvideo_args.pipeline_config is not None:
# Basic validation or usage if needed, otherwise just pass
pass
def create_pipeline_stages(self, fastvideo_args: FastVideoArgs) -> None:
self.add_stage(
stage_name="input_validation_stage",
@@ -56,8 +48,15 @@ class GlmImagePipeline(LoRAPipeline, ComposedPipelineBase):
processor=self.get_module("processor"),
transformer=self.get_module("transformer"),
scheduler=self.get_module("scheduler"),
vision_language_encoder=self.get_module(
"vision_language_encoder"),
vision_language_encoder=self.get_module("vision_language_encoder"),
),
)
self.add_stage(
stage_name="glm_image_condition_encoding_stage",
stage=GlmImageConditionEncodingStage(
vae=self.get_module("vae"),
transformer=self.get_module("transformer"),
),
)
@@ -0,0 +1,14 @@
# SPDX-License-Identifier: Apache-2.0
"""GLM-Image pipeline stages."""
from fastvideo.pipelines.basic.glm_image.stages.before_denoising import (GlmImageBeforeDenoisingStage)
from fastvideo.pipelines.basic.glm_image.stages.condition_encoding import (GlmImageConditionEncodingStage)
from fastvideo.pipelines.basic.glm_image.stages.decoding import (GlmImageDecodingStage)
from fastvideo.pipelines.basic.glm_image.stages.denoising import (GlmImageDenoisingStage)
__all__ = [
"GlmImageBeforeDenoisingStage",
"GlmImageConditionEncodingStage",
"GlmImageDecodingStage",
"GlmImageDenoisingStage",
]
@@ -0,0 +1,266 @@
# SPDX-License-Identifier: Apache-2.0
from __future__ import annotations
import re
from math import sqrt
import numpy as np
import torch
from fastvideo.distributed import get_local_torch_device
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.logger import init_logger
from fastvideo.pipelines.pipeline_batch_info import ForwardBatch
from fastvideo.pipelines.stages.base import PipelineStage
from fastvideo.pipelines.stages.validators import StageValidators as V
from fastvideo.pipelines.stages.validators import VerificationResult
logger = init_logger(__name__)
def calculate_shift(
image_seq_len: int,
base_seq_len: int = 256,
base_shift: float = 0.25,
max_shift: float = 0.75,
) -> float:
return (image_seq_len / base_seq_len)**0.5 * max_shift + base_shift
def get_glyph_texts(prompt: str | list[str]) -> list[str] | list[list[str]]:
if isinstance(prompt, str):
prompts: list[str] = [prompt]
is_batch = False
else:
prompts = prompt
is_batch = True
out: list[list[str]] = []
for p in prompts:
out.append(
re.findall(r"'([^']*)'", p) + re.findall(r"“([^“”]*)”", p) + re.findall(r'"([^"]*)"', p) +
re.findall(r"「([^「」]*)」", p))
return out if is_batch else out[0]
def compute_glyph_embeds(
prompts: list[str],
tokenizer,
text_encoder,
device: torch.device,
dtype: torch.dtype,
max_sequence_length: int = 2048,
) -> torch.Tensor:
all_glyph_texts = get_glyph_texts(prompts)
all_glyph_embeds = []
for glyph_texts in all_glyph_texts:
if len(glyph_texts) == 0:
glyph_texts = [""]
input_ids = tokenizer(
glyph_texts,
max_length=max_sequence_length,
truncation=True,
).input_ids
input_ids = [[tokenizer.pad_token_id] * ((len(input_ids) + 1) % 2) + ids for ids in input_ids]
max_length = max(len(ids) for ids in input_ids)
attention_mask = torch.tensor(
[[1] * len(ids) + [0] * (max_length - len(ids)) for ids in input_ids],
device=device,
)
input_ids_t = torch.tensor(
[ids + [tokenizer.pad_token_id] * (max_length - len(ids)) for ids in input_ids],
device=device,
)
outputs = text_encoder(input_ids_t, attention_mask=attention_mask)
glyph_embeds = outputs.last_hidden_state[attention_mask.bool()].unsqueeze(0)
all_glyph_embeds.append(glyph_embeds)
max_seq_len = max(emb.size(1) for emb in all_glyph_embeds)
padded = []
for emb in all_glyph_embeds:
if emb.size(1) < max_seq_len:
pad = torch.zeros(emb.size(0), max_seq_len - emb.size(1), emb.size(2), device=device, dtype=emb.dtype)
emb = torch.cat([pad, emb], dim=1)
padded.append(emb)
return torch.cat(padded, dim=0).to(device=device, dtype=dtype)
def _grid_dims(height: int, width: int) -> tuple[int, int, int, int]:
th, tw = height // 32, width // 32
ratio = th / tw
pth = int(sqrt(ratio) * 16)
ptw = int(sqrt(1 / ratio) * 16)
return th, tw, pth, ptw
def _upsample_d32_to_d16(tokens: torch.Tensor, th: int, tw: int) -> torch.Tensor:
tokens = tokens.view(1, 1, th, tw).float()
tokens = torch.nn.functional.interpolate(tokens, scale_factor=2, mode="nearest").long()
return tokens.view(1, -1)
class GlmImageBeforeDenoisingStage(PipelineStage):
def __init__(self,
vae,
text_encoder,
tokenizer,
processor,
transformer,
scheduler,
vision_language_encoder=None) -> None:
super().__init__()
self.vae = vae
self.text_encoder = text_encoder
self.tokenizer = tokenizer
self.processor = processor
self.transformer = transformer
self.scheduler = scheduler
if isinstance(vision_language_encoder, tuple):
self.vision_language_encoder, self.vl_processor = (vision_language_encoder[0], vision_language_encoder[1]
or processor)
else:
self.vision_language_encoder = vision_language_encoder
self.vl_processor = processor
@torch.no_grad()
def forward(self, batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> ForwardBatch:
device = get_local_torch_device()
dtype = torch.bfloat16
th, tw, pth, ptw = _grid_dims(batch.height, batch.width)
if batch.seed is not None:
torch.manual_seed(batch.seed)
if torch.cuda.is_available():
torch.cuda.manual_seed(batch.seed)
# 1-3. AR token generation. I2I prepends the condition image and uses a
# single-scale target grid; T2I is multi-scale.
is_t2i = batch.pil_image is None
if self.vision_language_encoder is not None:
content = [{"type": "text", "text": batch.prompt}]
if not is_t2i:
content.insert(0, {"type": "image", "image": batch.pil_image})
messages = [{"role": "user", "content": content}]
inputs = self.vl_processor.apply_chat_template(messages,
tokenize=True,
target_h=batch.height,
target_w=batch.width,
return_dict=True,
return_tensors="pt").to(device)
if is_t2i:
up_h, up_w = th, tw
large_start, large_count = pth * ptw, th * tw
max_new = large_count + (pth * ptw) + 1
else:
# Condition grid(s) first, target grid last.
_, t_h, t_w = inputs["image_grid_thw"][-1].tolist()
up_h, up_w = int(t_h), int(t_w)
large_start, large_count = 0, up_h * up_w
max_new = large_count + 1
outputs = self.vision_language_encoder.generate(**inputs, max_new_tokens=max_new, do_sample=True)
gen_tokens = outputs[0][inputs.input_ids.shape[-1]:]
if len(gen_tokens) >= large_start + large_count:
large_tokens = gen_tokens[large_start:large_start + large_count]
else:
available = gen_tokens[large_start:] if len(gen_tokens) > large_start else gen_tokens
large_tokens = torch.zeros(large_count, dtype=gen_tokens.dtype, device=gen_tokens.device)
large_tokens[:min(len(available), large_count)] = available[:large_count]
logger.warning("AR generated %d tokens, expected %d. Padding with zeros.", len(gen_tokens),
large_start + large_count)
batch.prior_token_id = _upsample_d32_to_d16(large_tokens, up_h, up_w)
batch.prior_token_drop = torch.zeros(batch.prior_token_id.shape, dtype=torch.bool, device=device)
if not is_t2i:
self._compute_source_prior_tokens(batch, inputs)
else:
num_prior_tokens = 4 * th * tw
logger.warning("No vision_language_encoder provided; using random dropped priors.")
batch.prior_token_id = torch.randint(0, 16384, (1, num_prior_tokens), device=device)
batch.prior_token_drop = torch.ones(batch.prior_token_id.shape, dtype=torch.bool, device=device)
# 4. Glyph T5 encoding.
prompts = [batch.prompt] if isinstance(batch.prompt, str) else list(batch.prompt)
prompt_embeds = compute_glyph_embeds(prompts, self.tokenizer, self.text_encoder, device, dtype)
# 5. CFG-side negative encoding.
if batch.do_classifier_free_guidance:
neg_prompts = [batch.negative_prompt or ""] * len(prompts)
neg_embeds = compute_glyph_embeds(neg_prompts, self.tokenizer, self.text_encoder, device, dtype)
L_pos, L_neg = prompt_embeds.shape[1], neg_embeds.shape[1]
max_L = max(L_pos, L_neg)
if L_pos < max_L:
pad = torch.zeros(prompt_embeds.shape[0],
max_L - L_pos,
prompt_embeds.shape[2],
device=device,
dtype=dtype)
prompt_embeds = torch.cat([pad, prompt_embeds], dim=1)
if L_neg < max_L:
pad = torch.zeros(neg_embeds.shape[0], max_L - L_neg, neg_embeds.shape[2], device=device, dtype=dtype)
neg_embeds = torch.cat([pad, neg_embeds], dim=1)
# Row 0 conditional (positive), row 1 unconditional (negative).
prompt_embeds = torch.cat([prompt_embeds, neg_embeds], dim=0)
att_pos = torch.ones((1, max_L), device=device)
att_neg = torch.ones((1, max_L), device=device)
if L_pos < max_L:
att_pos[:, :max_L - L_pos] = 0
if L_neg < max_L:
att_neg[:, :max_L - L_neg] = 0
attention_mask = torch.cat([att_pos, att_neg], dim=0)
else:
attention_mask = torch.ones((1, prompt_embeds.shape[1]), device=device)
batch.prompt_embeds = [prompt_embeds]
batch.attention_mask = attention_mask
# 6. Latents + dynamic flow shift.
if batch.seed is not None:
torch.manual_seed(batch.seed)
if torch.cuda.is_available():
torch.cuda.manual_seed(batch.seed)
batch.latents = torch.randn((1, 16, 1, batch.height // 8, batch.width // 8), device=device, dtype=dtype)
# Integer-cast linspace timesteps with resolution-dependent shift applied to
# sigmas only; the DiT is conditioned on the unshifted integer timesteps.
ntt = self.scheduler.config.num_train_timesteps
patch_size = self.transformer.patch_size
image_seq_len = ((batch.height // 8) * (batch.width // 8)) // (patch_size**2)
sched_timesteps = np.linspace(ntt, 1.0, batch.num_inference_steps + 1)[:-1].astype(np.int64).astype(np.float32)
sched_sigmas = sched_timesteps / ntt
self.scheduler.set_shift(calculate_shift(image_seq_len))
self.scheduler.set_timesteps(batch.num_inference_steps,
device=device,
sigmas=sched_sigmas.tolist(),
timesteps=sched_timesteps.tolist())
batch.timesteps = self.scheduler.timesteps
return batch
@torch.no_grad()
def _compute_source_prior_tokens(self, batch: ForwardBatch, inputs) -> None:
image_grid_thw = inputs["image_grid_thw"]
num_condition_images = image_grid_thw.shape[0] - 1
source_grids = image_grid_thw[:num_condition_images]
embed = torch.cat(self.vision_language_encoder.get_image_features(inputs["pixel_values"], source_grids), dim=0)
src_ids_d32 = self.vision_language_encoder.get_image_tokens(embed, source_grids)
split_sizes = source_grids.prod(dim=-1).tolist()
upsampled = [
_upsample_d32_to_d16(ids, int(grid[1]), int(grid[2])).squeeze(0)
for ids, grid in zip(torch.split(src_ids_d32, split_sizes), source_grids, strict=False)
]
src_grids_up = source_grids.clone()
src_grids_up[:, 1] *= 2
src_grids_up[:, 2] *= 2
batch.extra["glm_prior_token_image_ids"] = torch.cat(upsampled, dim=0)
batch.extra["glm_source_image_grid_thw"] = src_grids_up
def verify_input(self, batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> VerificationResult:
result = VerificationResult()
result.add_check("prompt", batch.prompt, V.string_not_empty)
return result
def verify_output(self, batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> VerificationResult:
result = VerificationResult()
result.add_check("prior_token_id", batch.prior_token_id, V.is_tensor)
return result
@@ -0,0 +1,65 @@
# SPDX-License-Identifier: Apache-2.0
from __future__ import annotations
import torch
from fastvideo.distributed import get_local_torch_device
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.forward_context import set_forward_context
from fastvideo.image_processor import ImageProcessor
from fastvideo.models.dits.glm_image import GlmImageKVCache
from fastvideo.pipelines.pipeline_batch_info import ForwardBatch
from fastvideo.pipelines.stages.base import PipelineStage
_CONDITION_MULTIPLE_OF = 16 # vae_scale_factor (8) * DiT patch_size (2)
class GlmImageConditionEncodingStage(PipelineStage):
def __init__(self, vae, transformer) -> None:
super().__init__()
self.vae = vae
self.transformer = transformer
self.image_processor = ImageProcessor(vae_scale_factor=_CONDITION_MULTIPLE_OF)
@torch.no_grad()
def forward(self, batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> ForwardBatch:
if batch.pil_image is None:
return batch
device = get_local_torch_device()
dtype = torch.bfloat16
self.vae.to(device)
prior_ids = batch.extra["glm_prior_token_image_ids"].to(device)
if prior_ids.dim() == 1:
prior_ids = prior_ids.unsqueeze(0)
# Latent patch count must match the source prior tokens; mismatch is fatal.
src_grid = batch.extra["glm_source_image_grid_thw"][0]
cond_h = int(src_grid[1]) * _CONDITION_MULTIPLE_OF
cond_w = int(src_grid[2]) * _CONDITION_MULTIPLE_OF
cond_img = self.image_processor.preprocess(batch.pil_image, cond_h, cond_w).to(device=device,
dtype=torch.float32)
latent = self.vae.encode(cond_img).latent_dist.mode()
cfg = self.vae.config
mean = torch.tensor(cfg.latents_mean, device=device, dtype=torch.float32).view(1, -1, 1, 1)
std = torch.tensor(cfg.latents_std, device=device, dtype=torch.float32).view(1, -1, 1, 1)
latent = ((latent - mean) / std).to(dtype)
kv_caches = GlmImageKVCache(num_layers=self.transformer.num_layers)
empty_text = batch.prompt_embeds[0][:1, :0, :].to(device=device, dtype=dtype)
with set_forward_context(current_timestep=0, attn_metadata=None, forward_batch=batch):
self.transformer(
hidden_states=latent,
encoder_hidden_states=empty_text,
prior_token_id=prior_ids,
prior_token_drop=torch.zeros((prior_ids.shape[0], ), dtype=torch.bool, device=device),
timestep=torch.zeros((1, ), device=device),
target_size=torch.tensor([tuple(cond_img.shape[-2:])], device=device, dtype=torch.long),
crop_coords=torch.zeros((1, 2), device=device, dtype=torch.long),
kv_caches=kv_caches,
kv_caches_mode="write",
)
batch.extra["glm_kv_caches"] = kv_caches
return batch
@@ -0,0 +1,35 @@
# SPDX-License-Identifier: Apache-2.0
from __future__ import annotations
import torch
from fastvideo.distributed import get_local_torch_device
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.pipelines.stages.decoding import DecodingStage
from fastvideo.utils import PRECISION_TO_TYPE
class GlmImageDecodingStage(DecodingStage):
@torch.no_grad()
def decode(self, latents: torch.Tensor, fastvideo_args: FastVideoArgs) -> torch.Tensor:
self.vae.to(get_local_torch_device())
latents = latents.to(get_local_torch_device())
vae_dtype = PRECISION_TO_TYPE[fastvideo_args.pipeline_config.vae_precision]
vae_autocast = (vae_dtype != torch.float32 and not fastvideo_args.disable_autocast)
latents = self._denormalize_latents(latents)
if latents.dim() == 5:
latents = latents.squeeze(2)
with torch.autocast(device_type="cuda", dtype=vae_dtype, enabled=vae_autocast):
if fastvideo_args.pipeline_config.vae_tiling:
self.vae.enable_tiling()
if not vae_autocast:
latents = latents.to(vae_dtype)
decoded = self.vae.decode(latents)
image = decoded.sample if hasattr(decoded, "sample") else decoded
image = (image / 2 + 0.5).clamp(0, 1)
return image.unsqueeze(2)
@@ -0,0 +1,192 @@
# SPDX-License-Identifier: Apache-2.0
"""CFG convention (both denoise paths): row 0 conditional (positive), row 1 unconditional."""
from __future__ import annotations
import torch
from fastvideo.attention.backends.sdpa import SDPAMetadata
from fastvideo.distributed import get_local_torch_device
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.forward_context import set_forward_context
from fastvideo.pipelines.pipeline_batch_info import ForwardBatch
from fastvideo.pipelines.stages.denoising import DenoisingStage
from fastvideo.pipelines.stages.validators import StageValidators as V
from fastvideo.pipelines.stages.validators import VerificationResult
from fastvideo.platforms import AttentionBackendEnum
class GlmImageDenoisingStage(DenoisingStage):
def verify_input(self, batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> VerificationResult:
result = VerificationResult()
result.add_check("timesteps", batch.timesteps, [V.is_tensor, V.min_dims(1)])
latents = getattr(batch, "latent", getattr(batch, "latents", None))
result.add_check("latents", latents, [V.is_tensor, V.with_dims(5)])
result.add_check("num_inference_steps", batch.num_inference_steps, V.positive_int)
result.add_check("prompt_embeds", batch.prompt_embeds, V.list_not_empty)
return result
def forward(self, batch: ForwardBatch, fastvideo_args: FastVideoArgs) -> ForwardBatch:
device = get_local_torch_device()
dtype = torch.bfloat16
guidance_scale = getattr(batch, "guidance_scale", 1.5)
do_cfg = guidance_scale > 1.0
latents = getattr(batch, "latent", getattr(batch, "latents", None))
if latents is None:
raise ValueError("No latents found in batch.")
if latents.dim() == 5:
latents = latents.squeeze(2)
prompt_embeds = batch.prompt_embeds[0]
text_attention_mask = getattr(batch, "attention_mask", None)
timesteps = batch.timesteps
patch_size = self.transformer.patch_size
_, _, h, w = latents.shape
image_seq_length = (h // patch_size) * (w // patch_size)
text_seq_length = prompt_embeds.shape[1] if prompt_embeds.dim() >= 2 else 0
first_block = self.transformer.transformer_blocks[0]
backend = getattr(first_block.attn1.attn, "backend", None)
sdpa = backend == AttentionBackendEnum.TORCH_SDPA and text_attention_mask is not None
kv_caches = batch.extra.get("glm_kv_caches")
if kv_caches is None:
self._denoise_t2i(batch, latents, prompt_embeds, text_attention_mask, timesteps, do_cfg, guidance_scale,
text_seq_length, image_seq_length, sdpa, device, dtype)
else:
self._denoise_i2i(batch, latents, prompt_embeds, text_attention_mask, timesteps, do_cfg, guidance_scale,
text_seq_length, image_seq_length, sdpa, kv_caches, device, dtype)
return batch
def _denoise_t2i(self, batch, latents, prompt_embeds, text_attention_mask, timesteps, do_cfg, guidance_scale,
text_seq_length, image_seq_length, sdpa, device, dtype):
num_inference_steps = batch.num_inference_steps
bs = 2 if do_cfg else 1
target_size = torch.tensor([[batch.height, batch.width]], device=device, dtype=torch.long).repeat(bs, 1)
crop_coords = torch.zeros((bs, 2), device=device, dtype=torch.long)
prior_token_id = batch.prior_token_id
if do_cfg and prior_token_id.shape[0] == 1:
prior_token_id = prior_token_id.repeat(2, 1)
if do_cfg:
prior_token_drop = torch.tensor([False, True], device=device)
else:
prior_token_drop = getattr(batch, "prior_token_drop", torch.tensor([False], device=device))
attention_mask_kv = None
if sdpa:
if (text_attention_mask.shape[0] == 1 and bs > 1):
text_attention_mask = text_attention_mask.repeat(bs, 1)
mix_attn_mask = torch.ones((bs, text_seq_length + image_seq_length), device=device, dtype=torch.float32)
mix_attn_mask[:, :text_seq_length] = (text_attention_mask.float().to(device))
attention_mask_kv = (mix_attn_mask > 0).unsqueeze(1).unsqueeze(2)
with self.progress_bar(total=num_inference_steps) as progress_bar:
for i, t in enumerate(timesteps):
latent_model_input = torch.cat([latents] * 2) if do_cfg else latents
latent_model_input = self.scheduler.scale_model_input(latent_model_input, t).to(dtype)
t_expand = t.expand(latent_model_input.shape[0]) - 1
attn_metadata = (SDPAMetadata(current_timestep=i, attn_mask=attention_mask_kv)
if attention_mask_kv is not None else None)
with torch.no_grad(), set_forward_context(current_timestep=i,
attn_metadata=attn_metadata,
forward_batch=batch):
noise_pred = self.transformer(
latent_model_input,
prompt_embeds,
prior_token_id,
prior_token_drop,
t_expand,
target_size,
crop_coords,
)
if do_cfg:
noise_pred_cond, noise_pred_uncond = noise_pred.float().chunk(2)
noise_pred = noise_pred_uncond + guidance_scale * (noise_pred_cond - noise_pred_uncond)
guidance_rescale = getattr(batch, "guidance_rescale", 0.0)
if guidance_rescale > 0.0:
dims = list(range(1, noise_pred_cond.ndim))
std_text = noise_pred_cond.std(dim=dims, keepdim=True)
std_cfg = noise_pred.std(dim=dims, keepdim=True)
rescaled = noise_pred * (std_text / std_cfg)
noise_pred = (guidance_rescale * rescaled + (1 - guidance_rescale) * noise_pred)
latents = self.scheduler.step(noise_pred, t, latents, return_dict=False)[0]
progress_bar.update()
batch.latents = latents.unsqueeze(2)
def _denoise_i2i(self, batch, latents, prompt_embeds, text_attention_mask, timesteps, do_cfg, guidance_scale,
text_seq_length, image_seq_length, sdpa, kv_caches, device, dtype):
"""Two separate transformer calls (cond reads the cache, uncond skips it):
the cache mode is one global flag with batch-1 k/v, so a 2-row CFG call
cannot express both."""
num_inference_steps = batch.num_inference_steps
target_size = torch.tensor([[batch.height, batch.width]], device=device, dtype=torch.long)
crop_coords = torch.zeros((1, 2), device=device, dtype=torch.long)
prior_token_id = batch.prior_token_id[:1]
drop_keep = torch.zeros((1, ), dtype=torch.bool, device=device)
drop_all = torch.ones((1, ), dtype=torch.bool, device=device)
cache_len = kv_caches[0].k_cache.shape[1]
def _mask(row: int, with_cache: bool):
if not sdpa:
return None
prefix = cache_len if with_cache else 0
m = torch.ones((1, prefix + text_seq_length + image_seq_length), device=device, dtype=torch.float32)
m[:, prefix:prefix + text_seq_length] = text_attention_mask[row:row + 1].float().to(device)
return (m > 0).unsqueeze(1).unsqueeze(2)
cond_mask = _mask(0, with_cache=True)
uncond_mask = _mask(1, with_cache=False) if do_cfg else None
pos_embeds = prompt_embeds[:1]
neg_embeds = prompt_embeds[1:2] if do_cfg else None
with self.progress_bar(total=num_inference_steps) as progress_bar:
for i, t in enumerate(timesteps):
latent_model_input = self.scheduler.scale_model_input(latents, t).to(dtype)
t_expand = t.expand(1) - 1
cond_meta = (SDPAMetadata(current_timestep=i, attn_mask=cond_mask) if cond_mask is not None else None)
with torch.no_grad(), set_forward_context(current_timestep=i,
attn_metadata=cond_meta,
forward_batch=batch):
noise_pred = self.transformer(latent_model_input,
pos_embeds,
prior_token_id,
drop_keep,
t_expand,
target_size,
crop_coords,
kv_caches=kv_caches,
kv_caches_mode="read")
if do_cfg:
uncond_meta = (SDPAMetadata(current_timestep=i, attn_mask=uncond_mask)
if uncond_mask is not None else None)
with torch.no_grad(), set_forward_context(current_timestep=i,
attn_metadata=uncond_meta,
forward_batch=batch):
noise_pred_uncond = self.transformer(latent_model_input,
neg_embeds,
prior_token_id,
drop_all,
t_expand,
target_size,
crop_coords,
kv_caches=kv_caches,
kv_caches_mode="skip")
noise_pred = noise_pred_uncond.float() + guidance_scale * (noise_pred.float() -
noise_pred_uncond.float())
latents = self.scheduler.step(noise_pred, t, latents, return_dict=False)[0]
progress_bar.update()
kv_caches.clear()
batch.latents = latents.unsqueeze(2)
-1
View File
@@ -16,7 +16,6 @@ from fastvideo.pipelines.lora_pipeline import LoRAPipeline
logger = init_logger(__name__)
_PREPROCESS_WORKLOAD_TYPE_TO_PIPELINE_NAME: dict[WorkloadType, str] = {
WorkloadType.I2V: "PreprocessPipelineI2V",
WorkloadType.T2V: "PreprocessPipelineT2V",
-5
View File
@@ -46,9 +46,6 @@ from fastvideo.pipelines.stages.longcat_video_vae_encoding import LongCatVideoVA
from fastvideo.pipelines.stages.longcat_kv_cache_init import LongCatKVCacheInitStage
from fastvideo.pipelines.stages.longcat_vc_denoising import LongCatVCDenoisingStage
# GLM-Image stages
from fastvideo.pipelines.stages.glm_image_before_denoising import GlmImageBeforeDenoisingStage
__all__ = [
"PipelineStage",
"InputValidationStage",
@@ -104,6 +101,4 @@ __all__ = [
"LongCatVideoVAEEncodingStage",
"LongCatKVCacheInitStage",
"LongCatVCDenoisingStage",
# GLM-Image stages
"GlmImageBeforeDenoisingStage",
]
@@ -1,345 +0,0 @@
# SPDX-License-Identifier: Apache-2.0
"""
GLM-Image before denoising stage.
This stage handles the preprocessing for GLM-Image generation, including:
- Prompt parsing and tokenization
- AR token generation via the vision-language encoder
- Latent preparation for the denoising process
"""
import os
import re
from math import sqrt
import torch
from fastvideo.distributed import get_local_torch_device
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.logger import init_logger
from fastvideo.pipelines.pipeline_batch_info import ForwardBatch
from fastvideo.pipelines.stages.base import PipelineStage
from fastvideo.pipelines.stages.validators import StageValidators as V
from fastvideo.pipelines.stages.validators import VerificationResult
logger = init_logger(__name__)
def calculate_shift(image_seq_len,
base_seq_len: int = 256,
base_shift: float = 0.25,
max_shift: float = 0.75) -> float:
m = (image_seq_len / base_seq_len)**0.5
return m * max_shift + base_shift
class GlmImageBeforeDenoisingStage(PipelineStage):
def __init__(self,
vae,
text_encoder,
tokenizer,
processor,
transformer,
scheduler,
vision_language_encoder=None):
super().__init__()
self.vae = vae
self.text_encoder = text_encoder
self.tokenizer = tokenizer
self.processor = processor
self.transformer = transformer
self.scheduler = scheduler
if isinstance(vision_language_encoder, tuple):
self.vision_language_encoder = vision_language_encoder[0]
self.vl_processor = vision_language_encoder[1] or processor
else:
self.vision_language_encoder = vision_language_encoder
self.vl_processor = processor
def _format_prompt(self, prompt: str, h: int, w: int) -> str:
"""Add <sop>H W<eop> to prompt if missing. H, W are pixel sizes / 32."""
th, tw = h // 32, w // 32
tag = f"<sop>{th} {tw}<eop>"
if tag not in prompt:
# Check if there's an existing tag to replace
prompt = re.sub(r"<sop>\d+\s+\d+<eop>", "", prompt)
prompt = f"{prompt} {tag}"
return prompt
def _parse_shape(self, prompt: str) -> tuple[int, int, int, int]:
match = re.search(r"<sop>(\d+)\s+(\d+)<eop>", prompt)
if not match:
raise ValueError(
f"Prompt must contain <sop>H W<eop>, got: {prompt}")
th, tw = int(match.group(1)), int(match.group(2))
ratio = th / tw
pth = int(sqrt(ratio) * 16)
ptw = int(sqrt(1 / ratio) * 16)
return th, tw, pth, ptw
def _upsample_tokens(self, tokens: torch.Tensor, th: int,
tw: int) -> torch.Tensor:
"""2x nearest neighbor upsample from d32 to d16 tokens."""
# tokens: [N]
tokens = tokens.view(1, 1, th, tw)
tokens = torch.nn.functional.interpolate(tokens.float(),
scale_factor=2,
mode="nearest").long()
return tokens.view(1, -1)
@torch.no_grad()
def forward(self, batch: ForwardBatch,
fastvideo_args: FastVideoArgs) -> ForwardBatch:
device = get_local_torch_device()
dtype = torch.bfloat16
# 1. Format and parse prompt
batch.prompt = self._format_prompt(batch.prompt, batch.height,
batch.width)
th, tw, pth, ptw = self._parse_shape(batch.prompt)
if batch.seed is not None:
torch.manual_seed(batch.seed)
if torch.cuda.is_available():
torch.cuda.manual_seed(batch.seed)
# 2. Generate AR tokens
if self.vision_language_encoder is not None:
# Deterministic comparison hook
if os.environ.get("FV_USE_SGL_PRIORS") == "1":
logger.warning(
"Injecting SGLang prior tokens for deterministic run")
# Load dumped priors from SGLang run
sgl_priors = torch.load(
"/workspace/debug_dumps/sgl/dit_input_prior_token_id_cond.pt",
map_location="cpu").to(device)
# These are already upsampled in SGLang dumps
batch.prior_token_id = sgl_priors
batch.prior_token_drop = torch.zeros(
batch.prior_token_id.shape, dtype=torch.bool, device=device)
else:
# Expand prompt for AR model (SGLang style)
expanded_prompt = batch.prompt.replace(
f"<sop>{th} {tw}<eop>",
f"<sop>{th} {tw}<eop><sop>{pth} {ptw}<eop>")
content = [{"type": "text", "text": expanded_prompt}]
if hasattr(batch, 'image') and batch.image is not None:
content.insert(0, {"type": "image", "image": batch.image})
messages = [{"role": "user", "content": content}]
inputs = self.vl_processor.apply_chat_template(
messages,
tokenize=True,
target_h=batch.height,
target_w=batch.width,
return_dict=True,
return_tensors="pt").to(device)
# Total tokens: small_image + large_image
max_new = (th * tw) + (pth * ptw) + 1
# Use sampling to match SGLang/creative generation quality
outputs = self.vision_language_encoder.generate(
**inputs,
max_new_tokens=max_new,
do_sample=True,
top_p=0.7,
top_k=50,
temperature=1.0)
# Extract large image tokens
gen_tokens = outputs[0][inputs.input_ids.shape[-1]:]
expected_start = pth * ptw
expected_count = th * tw
# Handle variable-length output: take what we can, pad if needed
if len(gen_tokens) >= expected_start + expected_count:
large_tokens = gen_tokens[expected_start:expected_start +
expected_count]
else:
# Fallback: pad with zeros if not enough tokens
available = gen_tokens[expected_start:] if len(
gen_tokens) > expected_start else gen_tokens
large_tokens = torch.zeros(expected_count,
dtype=gen_tokens.dtype,
device=gen_tokens.device)
large_tokens[:min(len(available), expected_count
)] = available[:expected_count]
logger.warning(
"AR generated %d tokens, expected %d. Padding with zeros.",
len(gen_tokens), expected_start + expected_count)
# Upsample to d16
batch.prior_token_id = self._upsample_tokens(
large_tokens, th, tw)
batch.prior_token_drop = torch.zeros(
batch.prior_token_id.shape,
dtype=torch.bool,
device=device)
else:
# No AR model - use random prior tokens (unconditional generation fallback)
# Number of patches after upsampling: (2*th) * (2*tw)
num_prior_tokens = 4 * th * tw # Upsampled from th*tw
logger.warning(
"No vision_language_encoder provided. Using random prior tokens for unconditional generation."
)
batch.prior_token_id = torch.randint(0,
16384, (1, num_prior_tokens),
device=device)
batch.prior_token_drop = torch.ones(
batch.prior_token_id.shape, dtype=torch.bool,
device=device) # Drop all priors
# 3. Encode prompts with T5 (Glyphs ONLY)
# SGLang logic: Only encode text inside quotes for rendering.
def get_glyph_texts(text):
text = text[0] if isinstance(text, list) else text
ocr_texts = (
re.findall(r"'([^']*)'", text)
+ re.findall(r"“([^“”]*)”", text)
+ re.findall(r'"([^"]*)"', text)
+ re.findall(r"「([^「」]*)」", text)
)
return ocr_texts
glyph_texts = get_glyph_texts(batch.prompt)
# If no glyphs, use empty string (or pad token)
texts_to_encode = glyph_texts if len(glyph_texts) > 0 else [""]
# Helper to encode list of texts and get embeddings matching SGLang style
# SGLang pads to max_length among the batch of glyphs, and masks rest.
# But here 'batch.prompt' is 1 string (BS=1 usually).
# We need to return [1, L, D] or [1, max_glyph_len, D]?
# SGLang: glyph_embeds = outputs.last_hidden_state[attention_mask.bool()].unsqueeze(0)
# It flattens valid tokens?
# Let's inspect SGLang `_get_glyph_embeds` (Step 627, lines 434-473)
# It creates a batch of glyphs. Encodes them.
# Then `outputs.last_hidden_state[attention_mask.bool()]` -> Selects ONLY valid tokens from all glyphs.
# .unsqueeze(0) -> [1, Total_Valid_Tokens, D]
text_inputs = self.tokenizer(
texts_to_encode,
padding="longest", # Pad to longest in this micro-batch
max_length=512,
truncation=True,
return_tensors="pt"
).to(device)
# Add padding for processing if needed (SGLang adds pad token buffering?)
# SGLang line 450: `[pad] * ... + input_ids` (Left/Right pad?)
# It seems extensive padding logic.
# For now, standard encoding + mask selection should approximate it.
prompt_outputs = self.text_encoder(
input_ids=text_inputs.input_ids,
attention_mask=text_inputs.attention_mask
)
# Flatten valid tokens from all glyphs into one sequence
# inputs.attention_mask identifies valid tokens (excluding pad)
valid_mask = text_inputs.attention_mask.bool()
prompt_embeds = prompt_outputs.last_hidden_state[valid_mask].unsqueeze(0).to(dtype)
# SGLang returns [1, Seq, D].
batch.attention_mask = None # Attention mask handled by packing?
# Actually SGLang `GlmImageAttention` uses `context_len` logic.
# FastVideo `GlmImageTransformerBlock` expects `encoder_hidden_states` [B, L, D].
# If L varies, it's fine.
# Handle Negatives (CFG)
if batch.do_classifier_free_guidance:
neg_prompt = batch.negative_prompt or ""
neg_glyph_texts = get_glyph_texts(neg_prompt)
neg_texts_to_encode = neg_glyph_texts if len(neg_glyph_texts) > 0 else [""]
neg_inputs = self.tokenizer(
neg_texts_to_encode,
padding="longest",
max_length=512,
truncation=True,
return_tensors="pt"
).to(device)
neg_outputs = self.text_encoder(
input_ids=neg_inputs.input_ids,
attention_mask=neg_inputs.attention_mask
)
neg_valid_mask = neg_inputs.attention_mask.bool()
neg_prompt_embeds = neg_outputs.last_hidden_state[neg_valid_mask].unsqueeze(0).to(dtype)
# Pad to match length for concatenation?
# Or use list? FastVideo handles lists?
# `glm_image.py` (Model) handles `encoder_hidden_states` as Tensor. I need to PAD.
L_pos = prompt_embeds.shape[1]
L_neg = neg_prompt_embeds.shape[1]
max_L = max(L_pos, L_neg)
# Pad with zeros
if L_pos < max_L:
pad = torch.zeros((1, max_L - L_pos, prompt_embeds.shape[2]), device=device, dtype=dtype)
prompt_embeds = torch.cat([prompt_embeds, pad], dim=1)
if L_neg < max_L:
pad = torch.zeros((1, max_L - L_neg, neg_prompt_embeds.shape[2]), device=device, dtype=dtype)
neg_prompt_embeds = torch.cat([neg_prompt_embeds, pad], dim=1)
prompt_embeds = torch.cat([neg_prompt_embeds, prompt_embeds], dim=0)
# Attention mask for transformer?
# We should provide explicit mask if we padded.
# SGLang: `text_attn_mask`.
# Here we construct mask: [1, max_L]
att_mask_pos = torch.ones((1, max_L), device=device)
if L_pos < max_L: att_mask_pos[:, L_pos:] = 0
att_mask_neg = torch.ones((1, max_L), device=device)
if L_neg < max_L: att_mask_neg[:, L_neg:] = 0
attention_mask = torch.cat([att_mask_neg, att_mask_pos], dim=0)
else:
attention_mask = torch.ones((1, prompt_embeds.shape[1]), device=device)
batch.prompt_embeds = [prompt_embeds]
batch.attention_mask = attention_mask
# 4. Prepare Latents
# GLM-Image uses 16 channels, VAE compression factor is 8
# Add extra dimension for video compatibility: [B, C, T, H, W] with T=1
# Set seed for deterministic latent generation
if batch.seed is not None:
torch.manual_seed(batch.seed)
if torch.cuda.is_available():
torch.cuda.manual_seed(batch.seed)
latents = torch.randn((1, 16, 1, batch.height // 8, batch.width // 8),
device=device,
dtype=dtype)
# Flow matching doesn't use init_noise_sigma, just pure noise
batch.latents = latents
# 5. Dynamic Flow Shift
num_patches = (batch.height // 16) * (batch.width // 16)
shift = calculate_shift(num_patches)
if hasattr(self.scheduler, 'set_shift'):
self.scheduler.set_shift(shift)
self.scheduler.set_timesteps(batch.num_inference_steps, device=device)
batch.timesteps = self.scheduler.timesteps
return batch
def verify_input(self, batch: ForwardBatch,
fastvideo_args: FastVideoArgs) -> VerificationResult:
result = VerificationResult()
result.add_check("prompt", batch.prompt, V.string_not_empty)
return result
def verify_output(self, batch: ForwardBatch,
fastvideo_args: FastVideoArgs) -> VerificationResult:
result = VerificationResult()
result.add_check("prior_token_id", batch.prior_token_id, V.is_tensor)
return result
@@ -1,62 +0,0 @@
# SPDX-License-Identifier: Apache-2.0
"""
GLM-Image decoding stage.
"""
import torch
from fastvideo.distributed import get_local_torch_device
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.models.vaes.common import ParallelTiledVAE
from fastvideo.pipelines.stages.decoding import DecodingStage
from fastvideo.utils import PRECISION_TO_TYPE
class GlmImageDecodingStage(DecodingStage):
"""Decoding stage for GLM-Image (2D image generation)."""
def __init__(self, vae: ParallelTiledVAE, pipeline=None) -> None:
super().__init__(vae, pipeline)
@torch.no_grad()
def decode(self, latents: torch.Tensor,
fastvideo_args: FastVideoArgs) -> torch.Tensor:
"""
Decode latent representations into pixel space using VAE.
Args:
latents: Input latent tensor with shape (batch, channels, frames, height_latents, width_latents)
fastvideo_args: Configuration containing VAE precision and tiling settings
Returns:
Decoded image tensor with shape (batch, channels, frames, height, width),
normalized to [0, 1] range
"""
self.vae.to(get_local_torch_device())
latents = latents.to(get_local_torch_device())
vae_dtype = PRECISION_TO_TYPE[
fastvideo_args.pipeline_config.vae_precision]
vae_autocast_enabled = (
vae_dtype != torch.float32) and not fastvideo_args.disable_autocast
print(f"DEBUG: Latents shape: {latents.shape}, dtype: {latents.dtype}, device: {latents.device}")
print(f"DEBUG: Latents stats: mean={latents.mean().item():.6f}, std={latents.std().item():.6f}, min={latents.min().item():.6f}, max={latents.max().item():.6f}")
latents = self._denormalize_latents(latents)
print(f"DEBUG: Denormalized latents stats: mean={latents.mean().item():.6f}, std={latents.std().item():.6f}, min={latents.min().item():.6f}, max={latents.max().item():.6f}")
if latents.dim() == 5:
latents = latents.squeeze(2)
with torch.autocast(device_type="cuda",
dtype=vae_dtype,
enabled=vae_autocast_enabled):
if fastvideo_args.pipeline_config.vae_tiling:
self.vae.enable_tiling()
if not vae_autocast_enabled:
latents = latents.to(vae_dtype)
decoded = self.vae.decode(latents)
image = decoded.sample if hasattr(decoded, 'sample') else decoded
image = (image / 2 + 0.5).clamp(0, 1)
image = image.unsqueeze(2)
return image
@@ -1,184 +0,0 @@
# SPDX-License-Identifier: Apache-2.0
"""
GLM-Image denoising stage.
This stage performs the denoising loop for GLM-Image generation.
"""
import torch
from fastvideo.fastvideo_args import FastVideoArgs
from fastvideo.forward_context import set_forward_context
from fastvideo.pipelines.pipeline_batch_info import ForwardBatch
from fastvideo.pipelines.stages.denoising import DenoisingStage
from fastvideo.distributed import get_local_torch_device
from fastvideo.pipelines.stages.validators import StageValidators as V
from fastvideo.pipelines.stages.validators import VerificationResult
from fastvideo.attention.backends.sdpa import SDPAMetadata
from fastvideo.platforms import AttentionBackendEnum
class GlmImageDenoisingStage(DenoisingStage):
"""Denoising stage for GLM-Image pipeline."""
def verify_input(self, batch: ForwardBatch,
fastvideo_args: FastVideoArgs) -> VerificationResult:
"""Verify denoising stage inputs."""
result = VerificationResult()
# Basic checks
result.add_check("timesteps", batch.timesteps,
[V.is_tensor, V.min_dims(1)])
# Use latents or latent
latents = getattr(batch, 'latent', getattr(batch, 'latents', None))
result.add_check("latents", latents, [V.is_tensor, V.with_dims(5)])
result.add_check("num_inference_steps", batch.num_inference_steps,
V.positive_int)
# We handle prompt_embeds as concatenated [neg, pos] for CFG
result.add_check("prompt_embeds", batch.prompt_embeds, V.list_not_empty)
# Explicitly skip negative_prompt_embeds logic from base class since we pre-concat
return result
def forward(self, batch: ForwardBatch,
fastvideo_args: FastVideoArgs) -> ForwardBatch:
device = get_local_torch_device()
dtype = torch.bfloat16
# Parameters
num_inference_steps = batch.num_inference_steps
guidance_scale = getattr(batch, 'guidance_scale', 1.5)
# Get latents from batch
latents = getattr(batch, 'latent', getattr(batch, 'latents', None))
if latents is None:
raise ValueError("No latents found in batch.")
# GLM-Image is 2D, squeeze temporal dimension if present [B, C, T, H, W] -> [B, C, H, W]
if latents.dim() == 5:
latents = latents.squeeze(2)
timesteps = batch.timesteps
prompt_embeds = batch.prompt_embeds[0] # (2, L, D) for CFG
text_attention_mask = getattr(batch, 'attention_mask', None)
# Prepare target_size and crop_coords
bs = 2 if guidance_scale > 1.0 else 1
target_size = torch.tensor([[batch.height, batch.width]],
device=device,
dtype=torch.long).repeat(bs, 1)
crop_coords = torch.zeros((bs, 2), device=device, dtype=torch.long)
# Prepare prior tokens for CFG
prior_token_id = batch.prior_token_id
if guidance_scale > 1.0 and prior_token_id.shape[0] == 1:
prior_token_id = prior_token_id.repeat(2, 1)
# Create drop mask: [False] for cond, [True] for uncond (batch dimension indexing)
prior_token_drop = torch.tensor([False, True], device=device)
else:
prior_token_drop = getattr(batch, 'prior_token_drop',
torch.tensor([False], device=device))
# Compute image sequence length from latents
# Get patch_size from transformer config (patch_size is in latent space)
patch_size = self.transformer.patch_size
_, _, h, w = latents.shape
image_seq_length = (h // patch_size) * (w // patch_size)
# Compute text sequence length from prompt_embeds
text_seq_length = prompt_embeds.shape[1] if prompt_embeds.dim(
) >= 2 else 0
# The mask will be set in forward context before each transformer call
with self.progress_bar(total=num_inference_steps) as progress_bar:
for i, t in enumerate(timesteps):
# Prepare model input (CFG pass)
if guidance_scale > 1.0:
latent_model_input = torch.cat([latents] * 2)
else:
latent_model_input = latents
# Scale model input for flow matching / diffusion
latent_model_input = self.scheduler.scale_model_input(
latent_model_input, t)
latent_model_input = latent_model_input.to(dtype)
t_expand = t.expand(latent_model_input.shape[0])
# Create combined attention mask for [text, image] sequence
attn_metadata = None
if text_attention_mask is not None:
first_block = self.transformer.transformer_blocks[0]
attn_backend = getattr(first_block.attn1.attn, 'backend',
None)
if attn_backend is not None and attn_backend == AttentionBackendEnum.TORCH_SDPA:
batch_size = latent_model_input.shape[0]
# Create combined mask: text part uses provided mask, image part is always 1
mix_attn_mask = torch.ones(
(batch_size, text_seq_length + image_seq_length),
device=device,
dtype=torch.float32)
# Expand text mask to batch size if needed
if text_attention_mask.shape[0] == 1 and batch_size > 1:
text_attention_mask = text_attention_mask.repeat(
batch_size, 1)
# Set text part of mask
mix_attn_mask[:, :
text_seq_length] = text_attention_mask.float(
).to(device)
# Convert to SDPA format: (B, 1, 1, L) for key-padding style mask
# True = attend, False = ignore (will be converted to additive mask by SDPA)
attention_mask_kv = (mix_attn_mask
> 0).unsqueeze(1).unsqueeze(
2) # (B, 1, 1, L)
# Create SDPAMetadata with the mask
attn_metadata = SDPAMetadata(
current_timestep=i, attn_mask=attention_mask_kv)
# If using Flash Attention, masks are not supported - attn_metadata remains None
with torch.no_grad(), set_forward_context(
current_timestep=i,
attn_metadata=attn_metadata,
forward_batch=batch):
# Predict noise
noise_pred = self.transformer(latent_model_input,
prompt_embeds, t_expand,
prior_token_id,
prior_token_drop, target_size,
crop_coords)
# Apply Classifier-Free Guidance
if guidance_scale > 1.0:
noise_pred_cond, noise_pred_uncond = noise_pred.chunk(2)
noise_pred = noise_pred_uncond + guidance_scale * (
noise_pred_cond - noise_pred_uncond)
# Apply guidance rescale if requested
guidance_rescale = getattr(batch, 'guidance_rescale',
0.0)
if guidance_rescale > 0.0:
std_text = noise_pred_cond.std(dim=list(
range(1, noise_pred_cond.ndim)),
keepdim=True)
std_cfg = noise_pred.std(dim=list(
range(1, noise_pred.ndim)),
keepdim=True)
noise_pred_rescaled = noise_pred * (std_text /
std_cfg)
noise_pred = guidance_rescale * noise_pred_rescaled + (
1 - guidance_rescale) * noise_pred
# Step scheduler
latents = self.scheduler.step(noise_pred,
t,
latents,
return_dict=False)[0]
progress_bar.update()
# Store result in batch (add temporal dim back for video-compatible decoding)
batch.latents = latents.unsqueeze(2) # [B, C, H, W] -> [B, C, 1, H, W]
return batch
+118 -92
View File
@@ -1,118 +1,144 @@
# SPDX-License-Identifier: Apache-2.0
"""SSIM-based regression tests for GLM-Image generation."""
"""SSIM-based regression test for GLM-Image generation."""
from __future__ import annotations
import os
from pathlib import Path
import pytest
import torch
from PIL import Image
import numpy as np
from fastvideo import VideoGenerator
from fastvideo.logger import init_logger
from fastvideo.tests.ssim.inference_similarity_utils import (
run_text_to_video_similarity_test,
)
from fastvideo.tests.ssim.reference_utils import (
get_cuda_device_name,
resolve_device_reference_folder,
)
logger = init_logger(__name__)
# Device-specific reference folder
device_name = torch.cuda.get_device_name()
if "A40" in device_name:
device_reference_folder = "A40_reference_videos"
elif "L40S" in device_name:
device_reference_folder = "L40S_reference_videos"
elif "H100" in device_name:
device_reference_folder = "H100_reference_videos"
else:
logger.warning(f"Unsupported device: {device_name}, using L40S references")
device_reference_folder = "L40S_reference_videos"
REQUIRED_GPUS = 1
# Test configuration
TEST_PROMPT = "A beautiful landscape photography with rolling hills, a winding river, and a vibrant sunset in the background. Warm golden light, photorealistic style."
REFERENCE_FILENAME = "landscape_ref.png"
MIN_ACCEPTABLE_SSIM = 0.98
REPO_ROOT = Path(__file__).resolve().parents[3]
LOCAL_WEIGHTS_DIR = Path(
os.getenv("GLM_IMAGE_LOCAL_WEIGHTS_DIR",
REPO_ROOT / "official_weights" / "glm_image"))
GLM_IMAGE_MODEL_PATH = os.getenv("GLM_IMAGE_MODEL_DIR", str(LOCAL_WEIGHTS_DIR))
device_reference_folder = resolve_device_reference_folder(
(
("A40", "A40"),
("L40S", "L40S"),
("H100", "H100"),
("H200", "H200"),
("B200", "B200"),
),
device_name=get_cuda_device_name(),
fallback_device_prefix="L40S",
logger=logger,
)
MODEL_ID = "zai-org__GLM-Image"
TEST_PROMPTS = [
"A beautiful landscape photography with rolling hills, "
"a winding river, and a vibrant sunset in the background. "
"Warm golden light, photorealistic style.",
]
GLM_IMAGE_PARAMS = {
"model_path": "zai-org/GLM-Image",
"num_gpus": 1,
"model_path": GLM_IMAGE_MODEL_PATH,
"sp_size": 1,
"tp_size": 1,
"height": 512,
"width": 512,
"num_frames": 1,
"fps": 1,
"num_inference_steps": 8,
"guidance_scale": 1.5,
"seed": 0,
"neg_prompt": "",
}
GLM_IMAGE_FULL_QUALITY_PARAMS = {
"num_gpus": 1,
"model_path": GLM_IMAGE_MODEL_PATH,
"sp_size": 1,
"tp_size": 1,
"height": 1024,
"width": 1024,
"num_frames": 1,
"fps": 1,
"num_inference_steps": 50,
"guidance_scale": 7.5,
"seed": 1024,
"guidance_scale": 1.5,
"seed": 0,
"neg_prompt": "",
}
GLM_IMAGE_MODEL_TO_PARAMS = {
MODEL_ID: GLM_IMAGE_PARAMS,
}
GLM_IMAGE_FULL_QUALITY_MODEL_TO_PARAMS = {
MODEL_ID: GLM_IMAGE_FULL_QUALITY_PARAMS,
}
def _compute_ssim(reference_path: str, generated_path: str) -> float:
"""Compute SSIM between two images."""
img1 = Image.open(reference_path).convert("RGB")
img2 = Image.open(generated_path).convert("RGB")
if img1.size != img2.size:
img2 = img2.resize(img1.size, Image.LANCZOS)
t1 = torch.from_numpy(np.array(img1)).permute(2, 0, 1).float() / 255.0
t2 = torch.from_numpy(np.array(img2)).permute(2, 0, 1).float() / 255.0
t1, t2 = t1.unsqueeze(0), t2.unsqueeze(0)
if torch.cuda.is_available():
t1, t2 = t1.cuda(), t2.cuda()
# Gaussian window for SSIM
window_size, sigma = 11, 1.5
coords = torch.arange(window_size, device=t1.device, dtype=t1.dtype) - (window_size - 1) / 2
g = torch.exp(-(coords**2) / (2 * sigma**2))
g = g / g.sum()
window = (g.view(-1, 1) @ g.view(1, -1)).view(1, 1, window_size, window_size)
window = window.expand(3, 1, window_size, window_size)
padding = window_size // 2
mu_x = torch.nn.functional.conv2d(t1, window, padding=padding, groups=3)
mu_y = torch.nn.functional.conv2d(t2, window, padding=padding, groups=3)
sigma_x2 = torch.nn.functional.conv2d(t1 * t1, window, padding=padding, groups=3) - mu_x * mu_x
sigma_y2 = torch.nn.functional.conv2d(t2 * t2, window, padding=padding, groups=3) - mu_y * mu_y
sigma_xy = torch.nn.functional.conv2d(t1 * t2, window, padding=padding, groups=3) - mu_x * mu_y
C1, C2 = 0.01**2, 0.03**2
ssim_map = ((2 * mu_x * mu_y + C1) * (2 * sigma_xy + C2)) / \
((mu_x**2 + mu_y**2 + C1) * (sigma_x2 + sigma_y2 + C2))
return float(ssim_map.mean().item())
def _has_weights() -> bool:
required = ["transformer", "vae", "text_encoder",
"vision_language_encoder", "processor", "tokenizer",
"scheduler"]
return all((LOCAL_WEIGHTS_DIR / r).exists() for r in required)
def test_glm_image_similarity():
"""Test GLM-Image output consistency against reference image."""
base_dir = os.path.dirname(os.path.abspath(__file__))
reference_path = os.path.join(base_dir, device_reference_folder, "glm_image", REFERENCE_FILENAME)
generated_dir = os.path.join(base_dir, "generated_images")
generated_path = os.path.join(generated_dir, REFERENCE_FILENAME)
os.makedirs(generated_dir, exist_ok=True)
generator = VideoGenerator.from_pretrained(
GLM_IMAGE_PARAMS["model_path"],
num_gpus=GLM_IMAGE_PARAMS["num_gpus"],
trust_remote_code=True,
)
def _upstream_glm_image_available() -> bool:
try:
result = generator.generate_video(
prompt=TEST_PROMPT,
output_path=generated_dir,
save_video=False,
return_frames=True,
height=GLM_IMAGE_PARAMS["height"],
width=GLM_IMAGE_PARAMS["width"],
num_inference_steps=GLM_IMAGE_PARAMS["num_inference_steps"],
guidance_scale=GLM_IMAGE_PARAMS["guidance_scale"],
seed=GLM_IMAGE_PARAMS["seed"],
)
import diffusers
import transformers
except ImportError:
return False
return (hasattr(transformers, "GlmImageForConditionalGeneration")
and hasattr(diffusers, "GlmImagePipeline"))
assert result and len(result) > 0, "No image generated"
Image.fromarray(result[0]).save(generated_path)
logger.info(f"Generated image saved to {generated_path}")
if not os.path.exists(reference_path):
pytest.skip(f"Reference image not found: {reference_path}")
ssim = _compute_ssim(reference_path, generated_path)
assert ssim >= MIN_ACCEPTABLE_SSIM, \
f"SSIM {ssim:.4f} below threshold {MIN_ACCEPTABLE_SSIM}"
logger.info(f"SSIM test passed: {ssim:.4f} >= {MIN_ACCEPTABLE_SSIM}")
finally:
generator.shutdown()
@pytest.mark.skipif(
not torch.cuda.is_available(),
reason="GLM-Image SSIM test requires CUDA",
)
@pytest.mark.skipif(
not _has_weights(),
reason=f"GLM-Image full weights not found at {LOCAL_WEIGHTS_DIR}.",
)
@pytest.mark.skipif(
not _upstream_glm_image_available(),
reason=("GLM-Image needs transformers>=5.0.0rc0 and "
"diffusers>=0.37.0.dev0; main pins predate both."),
)
@pytest.mark.parametrize("prompt", TEST_PROMPTS)
@pytest.mark.parametrize("attention_backend_name", ["TORCH_SDPA"])
@pytest.mark.parametrize("model_id", list(GLM_IMAGE_MODEL_TO_PARAMS.keys()))
def test_glm_image_similarity(
prompt: str,
attention_backend_name: str,
model_id: str,
) -> None:
run_text_to_video_similarity_test(
logger=logger,
script_dir=os.path.dirname(os.path.abspath(__file__)),
device_reference_folder=device_reference_folder,
prompt=prompt,
attention_backend_name=attention_backend_name,
model_id=model_id,
default_params_map=GLM_IMAGE_MODEL_TO_PARAMS,
full_quality_params_map=GLM_IMAGE_FULL_QUALITY_MODEL_TO_PARAMS,
min_acceptable_ssim=0.98,
init_kwargs_override={
"trust_remote_code": True,
"use_fsdp_inference": False,
},
generation_kwargs_override={
"save_video": True,
},
)
+191
View File
@@ -0,0 +1,191 @@
# GLM-Image Local Tests
Local-only parity and smoke tests for the `glm_image` FastVideo port. These
tests compare FastVideo against the official HF `transformers` + `diffusers`
GLM-Image implementation and are not expected to run in CI unless explicitly
promoted later.
Port progress, open questions, issues, and handoff notes live in
`tests/local_tests/glm_image/PORT_STATUS.md`.
## Reference Assets
| Field | Value |
|---|---|
| Model family | `glm_image` |
| Workload types | `T2I` (primary), `I2I` (planned follow-up) |
| Official reference | HF `transformers.models.glm_image` + HF `diffusers.pipelines.glm_image` |
| Local reference dir | `/home/hal-kaiqin/dreamverse/FastVideo-PR/reference/transformers/src/transformers/models/glm_image/`, `/home/hal-kaiqin/dreamverse/FastVideo-PR/reference/diffusers/src/diffusers/pipelines/glm_image/`, `/home/hal-kaiqin/dreamverse/FastVideo-PR/reference/diffusers/src/diffusers/models/transformers/transformer_glm_image.py` |
| Official commit/version | transformers `a65bf6c9d03322c1bf1963ef7c64289c4b3f0757`, diffusers `ff3b86b4755b46a7b5656dfcf84d25bd25ad4740` |
| HF weights | `zai-org/GLM-Image` |
| HF revision | default (`main`) |
| Local weights dir | `/home/hal-kaiqin/dreamverse/FastVideo-PR/official_weights/glm_image` (downloaded) |
| Source layout | `diffusers` |
| Needs conversion | `no` (HF repo already in diffusers component layout) |
Token env var (set externally if needed for gated downloads): `HF_TOKEN`.
The repo is currently public; no token required for inspect / public read.
## Shared Environment Setup
Run from the FastVideo repo root in the same conda/env used for FastVideo.
Do not create a separate upstream environment for parity tests.
Sparse-checkout HF reference repos (already done during prep):
```bash
git clone --depth 1 --filter=blob:none --sparse https://github.com/huggingface/transformers reference/transformers
git -C reference/transformers sparse-checkout init --cone
git -C reference/transformers sparse-checkout set src/transformers/models/glm_image
git clone --depth 1 --filter=blob:none --sparse https://github.com/huggingface/diffusers reference/diffusers
git -C reference/diffusers sparse-checkout init --cone
git -C reference/diffusers sparse-checkout set \
src/diffusers/pipelines/glm_image \
src/diffusers/models/transformers \
src/diffusers/models/autoencoders \
src/diffusers/schedulers
```
Editable installs are intentionally **not** used here: parity tests will import
`transformers.models.glm_image` and `diffusers.pipelines.glm_image` from the
already-installed HF packages on the active env, while the cloned reference
trees serve as source-readable parity references (Phase 1 architecture study,
key/shape dumps).
**`pyproject.toml` pins `transformers==5.0.0rc3`** — the first release shipping
GLM-Image's AR encoder (`GlmImageForConditionalGeneration`). So the AR model and
the full T2I/I2I pipeline run on the committed pins; no local override is needed.
Diffusers: production uses FastVideo's native DiT + a stock `AutoencoderKL`, so
`diffusers>=0.33.1` is enough. The diffusers `GlmImageTransformer2DModel` /
`GlmImagePipeline` classes are used **only by the parity tests** and need
`diffusers>=0.37.0.dev0`; those tests skip with actionable messages on older
diffusers (install it locally to run them):
```bash
uv pip install -U "git+https://github.com/huggingface/diffusers.git@ff3b86b4"
```
| Class | Used in | First available in |
|---|---|---|
| `transformers.GlmImageForConditionalGeneration` | production (AR encoder) | `transformers>=5.0.0rc0` — **the committed pin** |
| `diffusers.GlmImageTransformer2DModel`, `GlmImagePipeline` | parity tests only | `diffusers>=0.37.0.dev0` |
Do not change core dependency versions (`torch`, `flash-attn`, `triton`, CUDA
packages) without explicit approval.
## Official Environment Status
```text
dependency_changes: pyproject pins transformers==5.0.0rc3 (committed); diffusers test-only classes installed locally
official_env_status: imports_ok
private_dep_stubs: none
blocked_on: none
```
## Weight Setup
Run from the repo root (`/home/hal-kaiqin/dreamverse/FastVideo-PR`):
```bash
python ".agents/skills/add-model-01-prep/scripts/download_hf_weights.py" \
"zai-org/GLM-Image" \
"/home/hal-kaiqin/dreamverse/FastVideo-PR/official_weights/glm_image"
```
Weights are ~30 GB (vision_language_encoder 4 shards, transformer 3 shards,
vae + text_encoder). **Downloaded** to
`/home/hal-kaiqin/dreamverse/FastVideo-PR/official_weights/glm_image`. Cache via
`HF_HOME` if `/` is small.
## Prototype And Conversion Artifacts
`needs_conversion=no` for the base layout: HF weights are already in diffusers
component subfolders that FastVideo's diffusers-style loaders can consume,
**after** native components exist with state-dict surfaces that match. Param
mapping work belongs in `fastvideo/configs/models/dits/glm_image.py`
(`param_names_mapping`) and the new native VAE / encoder configs, not in a
conversion script.
```text
official_key_dumps:
transformer: converted_weights/glm_image/_mapping/transformer_official_keys.json
vae: converted_weights/glm_image/_mapping/vae_official_keys.json
text_encoder: converted_weights/glm_image/_mapping/text_encoder_official_keys.json
vision_language_encoder: converted_weights/glm_image/_mapping/vision_language_encoder_official_keys.json
fastvideo_key_dumps:
transformer: converted_weights/glm_image/_mapping/transformer_fastvideo_keys.json
vae: converted_weights/glm_image/_mapping/vae_fastvideo_keys.json
text_encoder: converted_weights/glm_image/_mapping/text_encoder_fastvideo_keys.json
vision_language_encoder: converted_weights/glm_image/_mapping/vision_language_encoder_fastvideo_keys.json
conversion_script: none (param_names_mapping in configs handles in-place rename at load)
conversion_source_layout: diffusers
converted_weights_dir: none
strict_load_status: not_run
```
## Expected Parity Tests
Local tests for this family (non-skip PASS on GB200; transformers 5.0.0rc3 — the
committed pin — plus diffusers 0.37.1 for the test-only reference classes):
| Component | Official files / args | Test | Concerns | Status |
|---|---|---|---|---|
| `transformer` (DiT) | `pipeline_glm_image.py::GlmImagePipeline.__init__` → `GlmImageTransformer2DModel` from `transformer/config.json`; defined in `transformer_glm_image.py` | `tests/local_tests/glm_image/test_transformer_parity.py` | Re-ported from diffusers ref. fp32 cosine 1.000000 (100% within 5e-3); bf16 cosine 0.999969 — the bf16 element-wise drift is rounding over 30 layers vs a different SDPA kernel, not a bug (I020). RoPE proven exactly equivalent. | `non_skip_pass` (fp32 + bf16 + strict-load) |
| `vae` (AutoencoderKL) | `pipeline_glm_image.py::GlmImagePipeline.__init__` → `AutoencoderKL` from `vae/config.json` | `tests/local_tests/glm_image/test_vae_parity.py` | Native `AutoencoderKL` under the lazy-wrapper exception file; config mirrors `vae/config.json` (block_out_channels, latents_mean/std). | `non_skip_pass` (decode + encode + config, 3/3) |
| `text_encoder` (T5) + `tokenizer` (**ByT5**) | `T5EncoderModel` + `ByT5Tokenizer` per `pipeline_glm_image.py:191-193` | `tests/local_tests/glm_image/test_t5_parity.py` | ByT5Tokenizer asserted; `compute_glyph_embeds` mirrors diffusers `_get_glyph_embeds` (Q003 answered — stock T5/ByT5 path works). | `non_skip_pass` (tokenizer + glyph extract + embeds, 3/3) |
| `vision_language_encoder` (GLM-4-9B AR) | `GlmImageForConditionalGeneration` from `modeling_glm_image.py`; instantiated via `from_pretrained(vision_language_encoder/)` | `tests/local_tests/glm_image/test_ar_parity.py` | Lazy-wrapper exception (E001) in `fastvideo/models/encoders/glm_image_ar_loader.py` confines the HF import. | `non_skip_pass` (surface + to/eval + greedy determinism, 3/3) |
| `processor` (`GlmImageProcessor`) | `transformers/models/glm_image/processing_glm_image.py` | covered alongside AR encoder | Loaded via `ProcessorLoader` (`AutoProcessor`, allowed utility). | `non_skip_pass` (via AR + pipeline) |
| `scheduler` (`FlowMatchEulerDiscreteScheduler`) | shared with other FastVideo flow-matching pipelines; `scheduler/scheduler_config.json` ships shift defaults | covered by pipeline parity test | Dynamic-shift schedule confirmed in practice by pipeline parity (I006). | `non_skip_pass` (via pipeline) |
| `pipeline` | `pipeline_glm_image.py::GlmImagePipeline.__call__` | `tests/local_tests/glm_image/test_pipeline_parity.py` | Pixel-exact parity is ill-posed (stochastic AR + independent latent RNG across codebases); test gates on FastVideo output validity + matching brightness regime vs diffusers. | `non_skip_pass` (smoke + cross-render, 2/2) |
| `pipeline` (I2I / edit) | `pipeline_glm_image.py::GlmImagePipeline.__call__` with `image=` (KV-cache write/read path) | `tests/local_tests/glm_image/test_edit_pipeline_parity.py` | Condition image conditions the DiT via a KV-cache write pass (not noise-the-latent). Validity/wiring gate; diffusers distribution parity deferred (same stochastic-AR caveat). Live-verified on GB200 (temp 5.0.0rc3 bump). | `non_skip_pass` (live validity); diffusers parity `deferred` |
Include reused components in this table. Reuse is accepted only after the
FastVideo component definition and official instantiation arguments have both
been checked and the component parity test passes non-skip.
Run the relevant tests with:
```bash
pytest tests/local_tests/glm_image/test_transformer_parity.py -v -s
pytest tests/local_tests/glm_image/test_vae_parity.py -v -s
pytest tests/local_tests/glm_image/test_t5_parity.py -v -s
pytest tests/local_tests/glm_image/test_ar_parity.py -v -s
pytest tests/local_tests/glm_image/test_pipeline_parity.py -v -s
pytest tests/local_tests/glm_image/test_edit_pipeline_parity.py -v -s
```
## Review Notes
- This is a **re-design** of an existing branch (`feature/glm-image-inference`).
The current branch contains a port adapted from SGLang with the following
defects that the re-design must fix:
1. Runtime `from transformers import GlmImageForConditionalGeneration` /
`AutoProcessor` / `AutoConfig` in `fastvideo/models/loader/component_loader.py`
and `from diffusers import AutoencoderKL` — violates production-boundary
rule.
2. Stray `print("DEBUG: …")` calls in
`fastvideo/pipelines/stages/glm_image_decoding.py`.
3. `FV_USE_SGL_PRIORS` debug branch in
`fastvideo/pipelines/stages/glm_image_before_denoising.py` that loads
dumped tensors from a hardcoded `/workspace/debug_dumps/sgl/...` path.
4. SSIM regression sits at **0.715** on L40S
(`fastvideo/tests/ssim/test_glm_image_similarity.py`) — too low to
function as a real quality bar.
5. T5 tokenizer is plain T5 instead of **ByT5** (diffusers reference uses
`ByT5Tokenizer`).
6. AR sampling temperature/top_p inconsistent between
`glm_image_before_denoising.py` (`top_p=0.7, top_k=50, temperature=1.0`)
and the pipeline config (`ar_top_p=0.75, ar_temperature=0.95`).
7. Pipeline stages live under `fastvideo/pipelines/stages/glm_image_*.py`
rather than `fastvideo/pipelines/basic/glm_image/stages/` per the
`add-model` Files Map.
8. Missing local parity tests (none under `tests/local_tests/glm_image/`,
`transformers/`, `vaes/`, `encoders/`, or `pipelines/`).
- Required before handoff: non-skip PASS for each required component parity
test, including reused components that own weights or numerical behavior.
- Pipeline parity may start as a scaffold, but final handoff requires non-skip
PASS or an explicit blocker accepted through the escape-hatch process.
- User decisions and pause points are tracked as `E###` rows in
`PORT_STATUS.md`; do not rely on chat history for escape-hatch context.
@@ -0,0 +1,109 @@
# SPDX-License-Identifier: Apache-2.0
from __future__ import annotations
import os
from pathlib import Path
import pytest
import torch
os.environ.setdefault("MASTER_ADDR", "localhost")
os.environ.setdefault("MASTER_PORT", "29519")
os.environ.setdefault("DISABLE_SP", "1")
REPO_ROOT = Path(__file__).resolve().parents[3]
FAMILY = "glm_image"
LOCAL_WEIGHTS_DIR = Path(
os.getenv("GLM_IMAGE_LOCAL_WEIGHTS_DIR",
REPO_ROOT / "official_weights" / FAMILY))
AR_DIR = LOCAL_WEIGHTS_DIR / "vision_language_encoder"
PROCESSOR_DIR = LOCAL_WEIGHTS_DIR / "processor"
def _has_weights() -> bool:
return AR_DIR.exists() and any(AR_DIR.glob("*.safetensors"))
def _has_glm_image_transformers() -> bool:
import importlib.util
spec = importlib.util.find_spec("transformers")
if spec is None:
return False
import transformers
return hasattr(transformers, "GlmImageForConditionalGeneration")
pytestmark = [
pytest.mark.skipif(
not _has_weights(),
reason=f"GLM-Image AR encoder weights not found at {AR_DIR}.",
),
pytest.mark.skipif(
not _has_glm_image_transformers(),
reason=("transformers in this env lacks "
"`GlmImageForConditionalGeneration` (main pins 4.57.3; needs "
"5.0.0rc0+). Bump transformers locally to exercise the AR "
"lazy-wrapper."),
),
]
SAMPLE_PROMPT = (
"A beautiful landscape photography with rolling hills, a winding river, "
"and a vibrant sunset in the background. Photorealistic style."
)
@pytest.fixture(scope="module")
def device() -> torch.device:
if not torch.cuda.is_available():
pytest.skip("CUDA required for GLM-Image AR encoder parity.")
return torch.device("cuda")
@pytest.fixture(scope="module")
def fastvideo_ar(device):
pytest.importorskip("fastvideo")
try:
from fastvideo.models.encoders.glm_image_ar_loader import (
GlmImageARLoader)
except ImportError as e:
pytest.skip(f"FastVideo lazy-wrapper not yet at target path: {e}")
loader = GlmImageARLoader(str(AR_DIR), str(PROCESSOR_DIR),
torch_dtype=torch.bfloat16)
loader.to(device).eval()
return loader
def test_ar_wrapper_exposes_expected_surface(fastvideo_ar):
assert callable(getattr(fastvideo_ar, "generate", None))
assert getattr(fastvideo_ar, "processor", None) is not None
assert getattr(fastvideo_ar, "config", None) is not None
assert getattr(fastvideo_ar, "generation_config", None) is not None
from transformers import GlmImageForConditionalGeneration
assert isinstance(fastvideo_ar._model, GlmImageForConditionalGeneration)
def test_ar_to_and_eval_propagate(fastvideo_ar, device):
assert next(fastvideo_ar._model.parameters()).device.type == device.type
assert not fastvideo_ar._model.training
def test_ar_generate_is_deterministic_under_do_sample_false(fastvideo_ar,
device):
processor = fastvideo_ar.processor
messages = [{
"role": "user",
"content": [{"type": "text", "text": SAMPLE_PROMPT}],
}]
inputs = processor.apply_chat_template(messages, tokenize=True,
target_h=1024, target_w=1024,
return_dict=True,
return_tensors="pt").to(device)
with torch.no_grad():
out_a = fastvideo_ar.generate(**inputs, max_new_tokens=32,
do_sample=False)
out_b = fastvideo_ar.generate(**inputs, max_new_tokens=32,
do_sample=False)
assert out_a.shape == out_b.shape
assert torch.equal(out_a, out_b)
@@ -0,0 +1,126 @@
# SPDX-License-Identifier: Apache-2.0
"""End-to-end scaffold for the GLM-Image image-to-image (edit) pipeline.
Drives FastVideo `VideoGenerator` in edit mode (a condition image is passed) and
checks that the unified pipeline routes through the condition-encoding stage +
KV-cache denoising path and emits a well-formed image. Like the T2I parity test,
pixel-exact parity vs diffusers is NOT asserted: the AR prior is sampled
(`do_sample=True`) and the diffusion latents draw from independent RNG streams,
so the codebases produce different valid samples of the same edit. We gate on
validity (and, when diffusers is present, a comparable brightness regime).
Skips cleanly until weights are available; GPU-heavy.
"""
from __future__ import annotations
import os
from pathlib import Path
import numpy as np
import pytest
import torch
from PIL import Image
os.environ.setdefault("MASTER_ADDR", "localhost")
os.environ.setdefault("MASTER_PORT", "29520")
os.environ.setdefault("DISABLE_SP", "1")
os.environ.setdefault("FASTVIDEO_ATTENTION_BACKEND", "TORCH_SDPA")
REPO_ROOT = Path(__file__).resolve().parents[3]
FAMILY = "glm_image"
LOCAL_WEIGHTS_DIR = Path(
os.getenv("GLM_IMAGE_LOCAL_WEIGHTS_DIR",
REPO_ROOT / "official_weights" / FAMILY))
CONDITION_IMAGE = REPO_ROOT / "assets" / "images" / "couple.jpg"
def _has_weights() -> bool:
required = ["transformer", "vae", "text_encoder",
"vision_language_encoder", "processor", "tokenizer",
"scheduler"]
return all((LOCAL_WEIGHTS_DIR / r).exists() for r in required)
def _upstream_glm_image_available() -> bool:
try:
import transformers
except ImportError:
return False
return hasattr(transformers, "GlmImageForConditionalGeneration")
pytestmark = [
pytest.mark.skipif(
not _has_weights(),
reason=f"GLM-Image full weights not found at {LOCAL_WEIGHTS_DIR}.",
),
pytest.mark.skipif(
not _upstream_glm_image_available(),
reason=("Edit pipeline needs transformers>=5.0.0rc0 "
"(GlmImageForConditionalGeneration); main pin predates it. "
"Bump locally to run."),
),
]
EDIT_PROMPT = "Make the scene a snowy winter landscape."
SEED = 0
HEIGHT = 512
WIDTH = 512
STEPS = 8 # low step count keeps the test under 5 min on a single GPU
@pytest.fixture(scope="module")
def device():
if not torch.cuda.is_available():
pytest.skip("CUDA required for GLM-Image edit pipeline.")
return torch.device("cuda")
def _to_uint8_hwc(a) -> np.ndarray:
"""Coerce a frame buffer (np or torch, [0,1] float or uint8, with optional
leading batch/time dims, HWC or CHW) to a single (H, W, 3) uint8 image."""
if torch.is_tensor(a):
a = a.float().cpu().numpy()
a = np.asarray(a)
while a.ndim > 3:
a = a[0]
if a.ndim == 3 and a.shape[0] in (1, 3) and a.shape[-1] not in (1, 3):
a = np.transpose(a, (1, 2, 0))
if a.dtype != np.uint8:
scale = 255.0 if float(a.max()) <= 1.5 else 1.0
a = np.clip(a * scale, 0, 255).astype(np.uint8)
return a
def _fastvideo_edit_image(device) -> np.ndarray:
pytest.importorskip("fastvideo")
try:
from fastvideo import VideoGenerator
except ImportError as e:
pytest.skip(f"FastVideo VideoGenerator unavailable: {e}")
condition = Image.open(CONDITION_IMAGE).convert("RGB")
gen = VideoGenerator.from_pretrained(str(LOCAL_WEIGHTS_DIR), num_gpus=1,
trust_remote_code=True)
result = gen.generate_video(prompt=EDIT_PROMPT,
pil_image=condition,
save_video=False,
return_frames=True,
height=HEIGHT, width=WIDTH,
num_inference_steps=STEPS,
guidance_scale=1.5,
seed=SEED)
gen.shutdown()
return _to_uint8_hwc(result["frames"][0])
def test_edit_pipeline_produces_valid_image(device):
"""Wiring gate for the edit path: passing a condition image routes through
the condition-encoding (KV-cache write) stage and the read/skip denoising
path, and the pipeline emits a well-formed, non-degenerate image of the
requested size. Component numerical correctness is covered by the (green)
DiT / VAE / T5 / AR component-parity tests."""
fv = _fastvideo_edit_image(device)
assert fv.shape == (HEIGHT, WIDTH, 3), f"unexpected shape {fv.shape}"
assert fv.dtype == np.uint8 and np.isfinite(fv).all()
assert fv.std() > 10.0, f"image is near-constant (std={fv.std():.2f})"
@@ -0,0 +1,138 @@
# SPDX-License-Identifier: Apache-2.0
from __future__ import annotations
import os
from pathlib import Path
import numpy as np
import pytest
import torch
os.environ.setdefault("MASTER_ADDR", "localhost")
os.environ.setdefault("MASTER_PORT", "29519")
os.environ.setdefault("DISABLE_SP", "1")
os.environ.setdefault("FASTVIDEO_ATTENTION_BACKEND", "TORCH_SDPA")
REPO_ROOT = Path(__file__).resolve().parents[3]
FAMILY = "glm_image"
LOCAL_WEIGHTS_DIR = Path(
os.getenv("GLM_IMAGE_LOCAL_WEIGHTS_DIR",
REPO_ROOT / "official_weights" / FAMILY))
def _has_weights() -> bool:
required = ["transformer", "vae", "text_encoder",
"vision_language_encoder", "processor", "tokenizer",
"scheduler"]
return all((LOCAL_WEIGHTS_DIR / r).exists() for r in required)
def _upstream_glm_image_available() -> bool:
try:
import transformers
import diffusers
except ImportError:
return False
return (hasattr(transformers, "GlmImageForConditionalGeneration")
and hasattr(diffusers, "GlmImagePipeline"))
pytestmark = [
pytest.mark.skipif(
not _has_weights(),
reason=f"GLM-Image full weights not found at {LOCAL_WEIGHTS_DIR}.",
),
pytest.mark.skipif(
not _upstream_glm_image_available(),
reason=("Pipeline parity needs transformers>=5.0.0rc0 and "
"diffusers>=0.37.0.dev0; main pins predate both. Bump locally "
"to run."),
),
]
SAMPLE_PROMPT = (
"A landscape photo with rolling green hills under a clear blue sky.")
SEED = 0
HEIGHT = 512
WIDTH = 512
STEPS = 8
@pytest.fixture(scope="module")
def device():
if not torch.cuda.is_available():
pytest.skip("CUDA required for GLM-Image pipeline parity.")
return torch.device("cuda")
def _to_uint8_hwc(a) -> np.ndarray:
if torch.is_tensor(a):
a = a.float().cpu().numpy()
a = np.asarray(a)
while a.ndim > 3:
a = a[0]
if a.ndim == 3 and a.shape[0] in (1, 3) and a.shape[-1] not in (1, 3):
a = np.transpose(a, (1, 2, 0))
if a.dtype != np.uint8:
scale = 255.0 if float(a.max()) <= 1.5 else 1.0
a = np.clip(a * scale, 0, 255).astype(np.uint8)
return a
def _diffusers_image(device) -> np.ndarray:
diffusers = pytest.importorskip("diffusers")
pipe = diffusers.GlmImagePipeline.from_pretrained(
str(LOCAL_WEIGHTS_DIR), torch_dtype=torch.bfloat16).to(device)
generator = torch.Generator(device=device).manual_seed(SEED)
out = pipe(prompt=SAMPLE_PROMPT, height=HEIGHT, width=WIDTH,
num_inference_steps=STEPS, guidance_scale=1.5,
generator=generator, output_type="np")
del pipe
torch.cuda.empty_cache()
return _to_uint8_hwc(out.images[0])
def _fastvideo_image(device) -> np.ndarray:
pytest.importorskip("fastvideo")
try:
from fastvideo import VideoGenerator
except ImportError as e:
pytest.skip(f"FastVideo VideoGenerator unavailable: {e}")
gen = VideoGenerator.from_pretrained(str(LOCAL_WEIGHTS_DIR), num_gpus=1,
trust_remote_code=True)
result = gen.generate_video(prompt=SAMPLE_PROMPT,
save_video=False,
return_frames=True,
height=HEIGHT, width=WIDTH,
num_inference_steps=STEPS,
guidance_scale=1.5,
seed=SEED)
gen.shutdown()
return _to_uint8_hwc(result["frames"][0])
def test_fastvideo_pipeline_produces_valid_image(device):
fv = _fastvideo_image(device)
assert fv.shape == (HEIGHT, WIDTH, 3), f"unexpected shape {fv.shape}"
assert fv.dtype == np.uint8 and np.isfinite(fv).all()
assert fv.std() > 10.0, f"image is near-constant (std={fv.std():.2f})"
def test_pipeline_matches_diffusers_distribution(device):
"""Pixel-exact parity is NOT asserted: AR sampling (do_sample=True) and
independent RNG streams make the two pipelines distinct valid samples, so we
only gate same-shape, healthy images in a comparable brightness regime."""
fv = _fastvideo_image(device)
official = _diffusers_image(device)
assert official.shape == fv.shape, (
f"shape mismatch: {official.shape} vs {fv.shape}")
assert official.std() > 10.0, "diffusers reference image is near-constant"
mae = np.abs(official.astype(np.float32) - fv.astype(np.float32)).mean()
bright_gap = abs(float(official.mean()) - float(fv.mean()))
print(f"[glm-image pipeline] MAE={mae:.2f}/255 "
f"brightness diffusers={official.mean():.1f} fv={fv.mean():.1f} "
f"gap={bright_gap:.1f}")
assert bright_gap < 60.0, (
f"global brightness regimes diverge ({bright_gap:.1f}/255); "
"suggests a wiring/normalization bug, not just RNG sampling")
@@ -0,0 +1,180 @@
# SPDX-License-Identifier: Apache-2.0
"""Parity scaffold for GLM-Image glyph T5 encoder.
The text_encoder is a small byte-level T5 (d_model=1472, gated-gelu, ByT5
tokenizer, vocab_size=384). This test exercises the glyph-only encoding path
that the diffusers pipeline implements as `_get_glyph_embeds` and compares
against the FastVideo glyph-encoding stage logic. Until the FastVideo stage is
refactored to mirror diffusers (per-prompt loop, even-length pad-prefix,
attention-mask flattening, left-padded batch), the parity test asserts the
high-impact pieces individually so failures point to the exact path that
diverged.
"""
from __future__ import annotations
import os
from pathlib import Path
import pytest
import torch
os.environ.setdefault("MASTER_ADDR", "localhost")
os.environ.setdefault("MASTER_PORT", "29519")
os.environ.setdefault("DISABLE_SP", "1")
REPO_ROOT = Path(__file__).resolve().parents[3]
FAMILY = "glm_image"
LOCAL_WEIGHTS_DIR = Path(
os.getenv("GLM_IMAGE_LOCAL_WEIGHTS_DIR",
REPO_ROOT / "official_weights" / FAMILY))
TEXT_ENCODER_DIR = LOCAL_WEIGHTS_DIR / "text_encoder"
TOKENIZER_DIR = LOCAL_WEIGHTS_DIR / "tokenizer"
def _has_weights() -> bool:
return (TEXT_ENCODER_DIR / "model.safetensors").exists() and TOKENIZER_DIR.exists()
pytestmark = pytest.mark.skipif(
not _has_weights(),
reason=f"GLM-Image text_encoder/tokenizer not found under {LOCAL_WEIGHTS_DIR}.",
)
SAMPLE_PROMPT = (
'A photo of a coffee shop with a wooden sign reading "Daily Grind" '
"and a chalk menu next to it"
)
@pytest.fixture(scope="module")
def device() -> torch.device:
if not torch.cuda.is_available():
pytest.skip("CUDA required for GLM-Image glyph encoder parity.")
return torch.device("cuda")
@pytest.fixture(scope="module")
def official_tokenizer():
transformers = pytest.importorskip("transformers")
return transformers.ByT5Tokenizer.from_pretrained(str(TOKENIZER_DIR))
@pytest.fixture(scope="module")
def official_text_encoder(device):
transformers = pytest.importorskip("transformers")
return transformers.T5EncoderModel.from_pretrained(
str(TEXT_ENCODER_DIR), torch_dtype=torch.float32).to(device).eval()
def test_tokenizer_is_byt5(official_tokenizer):
"""Glyph tokenizer must be ByT5 (byte-level), not generic T5."""
assert type(official_tokenizer).__name__ == "ByT5Tokenizer"
assert official_tokenizer.vocab_size <= 400, (
"ByT5 vocab is byte-level (~256+special)")
def test_glyph_text_extraction():
"""Glyph extraction must pull single+double+CJK quoted spans."""
pytest.importorskip("fastvideo")
try:
from fastvideo.pipelines.basic.glm_image.stages.before_denoising import ( # noqa
get_glyph_texts)
except ImportError as e:
pytest.skip(f"FastVideo glyph extractor not yet at target path: {e}")
out = get_glyph_texts(SAMPLE_PROMPT)
assert "Daily Grind" in out
def _diffusers_reference_get_glyph_embeds(prompts, tokenizer, text_encoder,
device, dtype,
max_sequence_length=2048):
"""Inlined copy of diffusers `_get_glyph_embeds` + `get_glyph_texts`.
Source: `reference/diffusers/src/diffusers/pipelines/glm_image/`
`pipeline_glm_image.py::GlmImagePipeline._get_glyph_embeds` at commit
ff3b86b4755b46a7b5656dfcf84d25bd25ad4740 (diffusers main, 0.37.0.dev0).
Inlined here because the installed diffusers package (`__version__`
typically <0.37.0) does not yet ship `GlmImagePipeline`. This function is
the reference oracle for FastVideo's `compute_glyph_embeds`.
"""
import re
def get_glyph_texts(prompt):
if isinstance(prompt, str):
prompt = [prompt]
out = []
for p in prompt:
out.append(
re.findall(r"'([^']*)'", p)
+ re.findall(r"“([^“”]*)”", p)
+ re.findall(r'"([^"]*)"', p)
+ re.findall(r"「([^「」]*)」", p))
return out
all_glyph_texts = get_glyph_texts(prompts)
all_glyph_embeds = []
for glyph_texts in all_glyph_texts:
if len(glyph_texts) == 0:
glyph_texts = [""]
input_ids = tokenizer(
glyph_texts,
max_length=max_sequence_length,
truncation=True,
).input_ids
input_ids = [
[tokenizer.pad_token_id] * ((len(input_ids) + 1) % 2) + ids
for ids in input_ids
]
max_length = max(len(ids) for ids in input_ids)
attention_mask = torch.tensor(
[[1] * len(ids) + [0] * (max_length - len(ids))
for ids in input_ids],
device=device,
)
input_ids_t = torch.tensor(
[
ids + [tokenizer.pad_token_id] * (max_length - len(ids))
for ids in input_ids
],
device=device,
)
outputs = text_encoder(input_ids_t, attention_mask=attention_mask)
glyph_embeds = outputs.last_hidden_state[attention_mask.bool()].unsqueeze(0)
all_glyph_embeds.append(glyph_embeds)
max_seq_len = max(emb.size(1) for emb in all_glyph_embeds)
padded = []
for emb in all_glyph_embeds:
if emb.size(1) < max_seq_len:
pad = torch.zeros(emb.size(0), max_seq_len - emb.size(1),
emb.size(2), device=device, dtype=emb.dtype)
emb = torch.cat([pad, emb], dim=1)
padded.append(emb)
return torch.cat(padded, dim=0).to(device=device, dtype=dtype)
def test_get_glyph_embeds_matches_diffusers(official_tokenizer,
official_text_encoder, device):
"""FastVideo `compute_glyph_embeds` must equal the diffusers reference."""
pytest.importorskip("fastvideo")
try:
from fastvideo.pipelines.basic.glm_image.stages.before_denoising import ( # noqa
compute_glyph_embeds)
except ImportError as e:
pytest.skip(f"FastVideo glyph encoder not yet at target path: {e}")
with torch.no_grad():
official_embeds = _diffusers_reference_get_glyph_embeds(
[SAMPLE_PROMPT], official_tokenizer, official_text_encoder, device,
torch.float32)
fv_embeds = compute_glyph_embeds([SAMPLE_PROMPT],
tokenizer=official_tokenizer,
text_encoder=official_text_encoder,
device=device,
dtype=torch.float32,
max_sequence_length=2048)
assert fv_embeds.shape == official_embeds.shape, (
f"shape mismatch: {fv_embeds.shape} vs {official_embeds.shape}")
torch.testing.assert_close(fv_embeds, official_embeds, atol=1e-4,
rtol=1e-4)
@@ -0,0 +1,205 @@
# SPDX-License-Identifier: Apache-2.0
"""Component parity scaffold for GLM-Image DiT (GlmImageTransformer2DModel).
Compares the FastVideo-native port at `fastvideo.models.dits.glm_image` against
the diffusers reference `GlmImageTransformer2DModel` from
`diffusers.models.transformers.transformer_glm_image`.
Skips cleanly until both the FastVideo class and the local weights exist; the
test is real (loads weights, forwards a fixed input, compares output tensors).
"""
from __future__ import annotations
import os
from pathlib import Path
import pytest
import torch
os.environ.setdefault("MASTER_ADDR", "localhost")
os.environ.setdefault("MASTER_PORT", "29519")
os.environ.setdefault("DISABLE_SP", "1")
os.environ.setdefault("FASTVIDEO_ATTENTION_BACKEND", "TORCH_SDPA")
REPO_ROOT = Path(__file__).resolve().parents[3]
FAMILY = "glm_image"
LOCAL_WEIGHTS_DIR = Path(
os.getenv("GLM_IMAGE_LOCAL_WEIGHTS_DIR",
REPO_ROOT / "official_weights" / FAMILY))
TRANSFORMER_DIR = LOCAL_WEIGHTS_DIR / "transformer"
def _has_weights() -> bool:
return TRANSFORMER_DIR.exists() and any(
TRANSFORMER_DIR.glob("*.safetensors"))
def _has_glm_image_diffusers() -> bool:
"""`GlmImageTransformer2DModel` lands in diffusers>=0.37.0.dev0.
FastVideo main pins `diffusers>=0.33.1`; the installed env is 0.36.0. The
numerical parity test needs the diffusers class as an oracle — until the
installed diffusers is upgraded, this test skips. The structural surface
is still verified by `test_glm_image_transformer_strict_load.py`.
"""
import importlib.util
spec = importlib.util.find_spec("diffusers")
if spec is None:
return False
import diffusers
return hasattr(diffusers, "GlmImageTransformer2DModel")
pytestmark = [
pytest.mark.skipif(
not _has_weights(),
reason=(
f"GLM-Image transformer weights not found at {TRANSFORMER_DIR}. "
"Download via "
"`python .agents/skills/add-model-01-prep/scripts/download_hf_weights.py "
"zai-org/GLM-Image official_weights/glm_image`."),
),
pytest.mark.skipif(
not _has_glm_image_diffusers(),
reason=("diffusers in this env lacks `GlmImageTransformer2DModel` "
"(main allows >=0.33.1; installed 0.36.0; class lands in "
"0.37.0.dev0). Bump diffusers locally to run numerical parity. "
"Structural surface is covered by "
"`test_glm_image_transformer_strict_load.py`."),
),
]
@pytest.fixture(scope="module")
def device() -> torch.device:
if not torch.cuda.is_available():
pytest.skip("CUDA required for GLM-Image transformer parity.")
return torch.device("cuda")
def _load_official(device, dtype):
diffusers = pytest.importorskip("diffusers")
cls = diffusers.GlmImageTransformer2DModel
return cls.from_pretrained(str(TRANSFORMER_DIR),
torch_dtype=dtype).to(device).eval()
def _load_fastvideo(device, dtype):
pytest.importorskip("fastvideo")
try:
from fastvideo.configs.models.dits.glm_image import GlmImageDiTConfig
from fastvideo.models.dits.glm_image import GlmImageTransformer2DModel
except ImportError as e:
pytest.skip(f"FastVideo GLM-Image DiT not yet ported: {e}")
cfg = GlmImageDiTConfig()
model = GlmImageTransformer2DModel(cfg,
{"_class_name":
"GlmImageTransformer2DModel"})
sd = _load_state_dict(TRANSFORMER_DIR)
sd = _apply_param_mapping(sd, cfg.arch_config.param_names_mapping)
missing, unexpected = model.load_state_dict(sd, strict=False)
assert not missing, f"FastVideo DiT missing keys: {missing[:10]}"
assert not unexpected, f"FastVideo DiT unexpected keys: {unexpected[:10]}"
return model.to(device, dtype=dtype).eval()
def _forward_pair(device, dtype):
"""Run both DiTs on identical synthetic inputs; return (fv_out, ref_out)."""
from fastvideo.forward_context import set_forward_context
ref = _load_official(device, dtype)
fv = _load_fastvideo(device, dtype)
inputs = _make_inputs(device, dtype=dtype)
with torch.no_grad():
ref_out = ref(**inputs, return_dict=False)[0]
with set_forward_context(current_timestep=0, attn_metadata=None,
forward_batch=None):
fv_out = fv(**inputs)
if isinstance(fv_out, tuple):
fv_out = fv_out[0]
assert ref_out.shape == fv_out.shape, (
f"shape mismatch: {ref_out.shape} vs {fv_out.shape}")
# Free the two 7B models before the next dtype runs (single-GPU budget).
del ref, fv
torch.cuda.empty_cache()
return fv_out.float(), ref_out.float()
def _load_state_dict(dir_path: Path) -> dict[str, torch.Tensor]:
safetensors = pytest.importorskip("safetensors.torch")
sd: dict[str, torch.Tensor] = {}
for shard in sorted(dir_path.glob("*.safetensors")):
sd.update(safetensors.load_file(str(shard)))
return sd
def _apply_param_mapping(sd, mapping):
import re
out = {}
for k, v in sd.items():
new_k = k
for pat, repl in mapping.items():
if re.match(pat, k):
new_k = re.sub(pat, repl, k)
break
out[new_k] = v
return out
def _make_inputs(device, batch_size=1, height=64, width=64,
text_seq_len=32, dtype=torch.bfloat16) -> dict[str, torch.Tensor]:
"""Synthetic inputs matching the diffusers `GlmImageTransformer2DModel.forward`
contract. Latent shape is `(B, 16, H, W)`; post-patch grid is `(H/2, W/2)`
so `prior_token_id` has `(H/2)*(W/2)` entries per batch element."""
torch.manual_seed(0)
patch_size = 2
num_patches = (height // patch_size) * (width // patch_size)
return {
"hidden_states": torch.randn(batch_size, 16, height, width,
device=device, dtype=dtype),
"encoder_hidden_states": torch.randn(batch_size, text_seq_len, 1472,
device=device, dtype=dtype),
"prior_token_id": torch.randint(0, 16384, (batch_size, num_patches),
device=device, dtype=torch.long),
"prior_token_drop": torch.zeros(batch_size, device=device,
dtype=torch.bool),
"timestep": torch.tensor([500] * batch_size, device=device,
dtype=torch.long),
"target_size": torch.tensor([[height * 8, width * 8]] * batch_size,
device=device, dtype=torch.long),
"crop_coords": torch.zeros(batch_size, 2, device=device,
dtype=torch.long),
}
# fp32 is the structural-correctness gate: with identical math the two
# implementations agree to ~1e-3 (residual is SDPA reduction-order noise).
ATOL_FP32 = 1e-2
RTOL_FP32 = 1e-2
# bf16 is the inference dtype. Rounding accumulates across 30 layers and the
# attention kernels differ, so element-wise atol is meaningless; cosine
# similarity is the right statistical bar for "same function, bf16 noise".
COS_MIN_BF16 = 0.999
def test_transformer_forward_matches_diffusers_fp32(device):
"""Structural gate: in fp32 the FastVideo DiT is numerically faithful to
the diffusers reference (cosine ~1.0, element-wise close to ~1e-3)."""
fv_out, ref_out = _forward_pair(device, torch.float32)
cos = torch.nn.functional.cosine_similarity(
fv_out.flatten(), ref_out.flatten(), dim=0)
assert cos > 0.9999, f"fp32 cosine similarity too low: {cos:.6f}"
torch.testing.assert_close(fv_out, ref_out,
atol=ATOL_FP32, rtol=RTOL_FP32)
def test_transformer_forward_matches_diffusers_bf16(device):
"""Inference-dtype sanity: bf16 output tracks the diffusers reference up to
precision accumulation (high cosine similarity). Element-wise atol is not a
meaningful bar for a 30-layer bf16 transformer vs a different attention
kernel (~28% of elements exceed atol=5e-3 purely from rounding) — see
PORT_STATUS I020."""
fv_out, ref_out = _forward_pair(device, torch.bfloat16)
cos = torch.nn.functional.cosine_similarity(
fv_out.flatten(), ref_out.flatten(), dim=0)
assert cos > COS_MIN_BF16, (
f"bf16 cosine similarity {cos:.6f} below {COS_MIN_BF16}; "
"indicates a structural divergence, not just precision.")
@@ -0,0 +1,94 @@
# SPDX-License-Identifier: Apache-2.0
"""Strict-load smoke for GLM-Image DiT.
Loads all three transformer shards from `official_weights/glm_image/transformer/`,
applies the FastVideo `param_names_mapping`, and calls
`GlmImageTransformer2DModel.load_state_dict(..., strict=True)`. Asserts the
config-driven model instantiates and every checkpoint key has a matching
FastVideo parameter with the same shape and dtype.
This is weight-load verification only — it does not run a forward pass. Numerical
parity against the diffusers reference lives in
`test_glm_image_transformer_parity.py` and requires `diffusers >= 0.37.0.dev0`.
"""
from __future__ import annotations
import os
import re
from pathlib import Path
import pytest
import torch
os.environ.setdefault("MASTER_ADDR", "localhost")
os.environ.setdefault("MASTER_PORT", "29519")
os.environ.setdefault("DISABLE_SP", "1")
os.environ.setdefault("FASTVIDEO_ATTENTION_BACKEND", "TORCH_SDPA")
REPO_ROOT = Path(__file__).resolve().parents[3]
FAMILY = "glm_image"
LOCAL_WEIGHTS_DIR = Path(
os.getenv("GLM_IMAGE_LOCAL_WEIGHTS_DIR",
REPO_ROOT / "official_weights" / FAMILY))
TRANSFORMER_DIR = LOCAL_WEIGHTS_DIR / "transformer"
def _has_shards() -> bool:
return TRANSFORMER_DIR.exists() and any(
TRANSFORMER_DIR.glob("*.safetensors"))
pytestmark = pytest.mark.skipif(
not _has_shards(),
reason=f"GLM-Image transformer shards not at {TRANSFORMER_DIR}.",
)
def test_dit_strict_load_against_hf_checkpoint():
safetensors = pytest.importorskip("safetensors.torch")
from fastvideo.configs.models.dits.glm_image import GlmImageDiTConfig
from fastvideo.models.dits.glm_image import GlmImageTransformer2DModel
cfg = GlmImageDiTConfig()
mapping = cfg.arch_config.param_names_mapping
# 1. Build the FastVideo model on CPU/meta-free path.
torch.set_default_device("cpu")
hf_config = {"_class_name": "GlmImageTransformer2DModel"}
model = GlmImageTransformer2DModel(cfg, hf_config)
# 2. Load every shard and apply param_names_mapping.
raw_sd: dict[str, torch.Tensor] = {}
for shard in sorted(TRANSFORMER_DIR.glob("*.safetensors")):
raw_sd.update(safetensors.load_file(str(shard)))
def rename(k: str) -> str:
for pat, repl in mapping.items():
if re.match(pat, k):
return re.sub(pat, repl, k)
return k
renamed_sd = {rename(k): v for k, v in raw_sd.items()}
# 3. Compare key sets (sanity, redundant with strict=True but gives a
# clearer error message if it fails).
fv_keys = set(model.state_dict().keys())
ckpt_keys = set(renamed_sd.keys())
missing = fv_keys - ckpt_keys
unexpected = ckpt_keys - fv_keys
assert not missing, f"FastVideo DiT missing {len(missing)} keys, e.g. {sorted(missing)[:5]}"
assert not unexpected, f"checkpoint has {len(unexpected)} unexpected keys, e.g. {sorted(unexpected)[:5]}"
# 4. Shape and dtype sanity — strict_load with shape mismatch raises early.
fv_sd = model.state_dict()
shape_mismatches = []
for k, v in renamed_sd.items():
if v.shape != fv_sd[k].shape:
shape_mismatches.append((k, tuple(fv_sd[k].shape), tuple(v.shape)))
assert not shape_mismatches, (
f"shape mismatches: {shape_mismatches[:5]}")
# 5. Strict load.
incompatible = model.load_state_dict(renamed_sd, strict=True)
assert not incompatible.missing_keys
assert not incompatible.unexpected_keys
@@ -0,0 +1,106 @@
# SPDX-License-Identifier: Apache-2.0
"""Component parity scaffold for GLM-Image VAE (AutoencoderKL).
Compares the FastVideo-native AutoencoderKL port against diffusers'
`AutoencoderKL` loaded from `zai-org/GLM-Image/vae`. The published checkpoint
uses `block_out_channels=[128, 512, 1024, 1024]`, `latent_channels=16`, and
per-channel `latents_mean[16]` + `latents_std[16]` normalization (not a scalar
`scaling_factor`).
Skips cleanly until both the FastVideo class and the VAE shard exist.
"""
from __future__ import annotations
import json
import os
from pathlib import Path
import pytest
import torch
os.environ.setdefault("MASTER_ADDR", "localhost")
os.environ.setdefault("MASTER_PORT", "29519")
os.environ.setdefault("DISABLE_SP", "1")
REPO_ROOT = Path(__file__).resolve().parents[3]
FAMILY = "glm_image"
LOCAL_WEIGHTS_DIR = Path(
os.getenv("GLM_IMAGE_LOCAL_WEIGHTS_DIR",
REPO_ROOT / "official_weights" / FAMILY))
VAE_DIR = LOCAL_WEIGHTS_DIR / "vae"
def _has_weights() -> bool:
return (VAE_DIR / "diffusion_pytorch_model.safetensors").exists()
pytestmark = pytest.mark.skipif(
not _has_weights(),
reason=f"GLM-Image VAE weights not found at {VAE_DIR}.",
)
@pytest.fixture(scope="module")
def device() -> torch.device:
if not torch.cuda.is_available():
pytest.skip("CUDA required for GLM-Image VAE parity.")
return torch.device("cuda")
@pytest.fixture(scope="module")
def vae_config():
with open(VAE_DIR / "config.json") as f:
return json.load(f)
@pytest.fixture(scope="module")
def official_vae(device):
diffusers = pytest.importorskip("diffusers")
vae = diffusers.AutoencoderKL.from_pretrained(
str(VAE_DIR), torch_dtype=torch.float32).to(device).eval()
return vae
@pytest.fixture(scope="module")
def fastvideo_vae(device):
pytest.importorskip("fastvideo")
try:
from fastvideo.configs.models.vaes.glm_image import GlmImageVAEConfig
from fastvideo.models.vaes.autoencoder_kl import AutoencoderKL
except ImportError as e:
pytest.skip(f"FastVideo native AutoencoderKL not yet ported: {e}")
cfg = GlmImageVAEConfig()
vae = AutoencoderKL(cfg)
safetensors = pytest.importorskip("safetensors.torch")
sd = safetensors.load_file(
str(VAE_DIR / "diffusion_pytorch_model.safetensors"))
missing, unexpected = vae.load_state_dict(sd, strict=False)
assert not missing, f"FastVideo VAE missing keys: {missing[:10]}"
assert not unexpected, f"FastVideo VAE unexpected keys: {unexpected[:10]}"
return vae.to(device, dtype=torch.float32).eval()
def test_vae_config_matches_real_checkpoint(vae_config):
assert vae_config["block_out_channels"] == [128, 512, 1024, 1024]
assert vae_config["latent_channels"] == 16
assert "latents_mean" in vae_config and len(
vae_config["latents_mean"]) == 16
assert "latents_std" in vae_config and len(vae_config["latents_std"]) == 16
def test_vae_decode_matches_diffusers(official_vae, fastvideo_vae, device):
torch.manual_seed(0)
latents = torch.randn(1, 16, 32, 32, device=device, dtype=torch.float32)
with torch.no_grad():
official_out = official_vae.decode(latents, return_dict=False)[0]
fv_out = fastvideo_vae.decode(latents, return_dict=False)[0]
torch.testing.assert_close(fv_out, official_out, atol=1e-3, rtol=1e-3)
def test_vae_encode_matches_diffusers(official_vae, fastvideo_vae, device):
torch.manual_seed(0)
pixels = torch.randn(1, 3, 256, 256, device=device, dtype=torch.float32)
with torch.no_grad():
official_z = official_vae.encode(pixels).latent_dist.mode()
fv_z = fastvideo_vae.encode(pixels).latent_dist.mode()
torch.testing.assert_close(fv_z, official_z, atol=1e-3, rtol=1e-3)