Compare commits

..
Author SHA1 Message Date
Roger Wang c11a317f22 update
Signed-off-by: Roger Wang <hey@rogerw.io>
2026-02-01 00:20:41 -08:00
Roger Wang 81774fd933 [Misc] Bump version to 0.14.0 (#1128)
Signed-off-by: Roger Wang <hey@rogerw.io>
2026-02-01 00:09:43 -08:00
汪志鹏andHongsheng Liu de0a8b057b [CI]: Bagel E2E Smoked Test (#1074)
Signed-off-by: princepride <wangzhipeng628@gmail.com>
Co-authored-by: Hongsheng Liu <liuhongsheng4@huawei.com>
2026-01-31 22:08:22 +13:00
汪志鹏andHongsheng Liu 0f0e94760c [TeaCache]: Add Coefficient Estimation (#940)
Signed-off-by: princepride <wangzhipeng628@gmail.com>
Signed-off-by: 汪志鹏 <wangzhipeng628@gmail.com>
Co-authored-by: Hongsheng Liu <liuhongsheng4@huawei.com>
2026-01-31 22:06:17 +13:00
13 changed files with 868 additions and 14 deletions

No files matched your search

+35
View File
@@ -305,3 +305,38 @@ steps:
hostPath:
path: /mnt/hf-cache
type: DirectoryOrCreate
- label: "Bagel Text2Img Model Test with H100"
timeout_in_minutes: 30
depends_on: image-build
commands:
- export VLLM_WORKER_MULTIPROC_METHOD=spawn
- pytest -s -v tests/e2e/offline_inference/test_bagel_text2img.py
agents:
queue: "mithril-h100-pool"
plugins:
- kubernetes:
podSpec:
containers:
- image: public.ecr.aws/q9t5s3a7/vllm-ci-test-repo:$BUILDKITE_COMMIT
resources:
limits:
nvidia.com/gpu: 1
volumeMounts:
- name: devshm
mountPath: /dev/shm
- name: hf-cache
mountPath: /root/.cache/huggingface
env:
- name: HF_HOME
value: /root/.cache/huggingface
nodeSelector:
node.kubernetes.io/instance-type: gpu-h100-sxm
volumes:
- name: devshm
emptyDir:
medium: Memory
- name: hf-cache
hostPath:
path: /mnt/hf-cache
type: DirectoryOrCreate
+1
View File
@@ -15,6 +15,7 @@ Easy, fast, and cheap omni-modality model serving for everyone
---
*Latest News* 🔥
- [2026/02] We released [0.14.0](https://github.com/vllm-project/vllm-omni/releases/tag/v0.14.0) - This is the first **stable release** of vLLM-Omni that expands Omni’s diffusion / image-video generation and audio / TTS stack, improves distributed execution and memory efficiency, and broadens platform/backend coverage (GPU/ROCm/NPU/XPU). It also brings meaningful upgrades to serving APIs, profiling & benchmarking, and overall stability.
- [2026/01] We released [0.12.0rc1](https://github.com/vllm-project/vllm-omni/releases/tag/v0.12.0rc1) - a major RC milestone focused on maturing the diffusion stack, strengthening OpenAI-compatible serving, expanding omni-model coverage, and improving stability across platforms (GPU/NPU/ROCm), please check our latest [design](https://docs.google.com/presentation/d/1qv4qMW1rKAqDREMXiUDLIgqqHQe7TDPj/edit?usp=sharing&ouid=110473603432222024453&rtpof=true&sd=true).
- [2025/11] vLLM community officially released [vllm-project/vllm-omni](https://github.com/vllm-project/vllm-omni) in order to support omni-modality models serving.
@@ -204,7 +204,151 @@ Key point for writing the example:
+ Save or display the generated results so users can validate the integration.
## Step 5: Open a Pull Request
## Step 5: TeaCache Coefficient Estimation (Optional)
If your model supports TeaCache acceleration, you need to estimate the polynomial coefficients for optimal caching performance.
### 5.1 Add Extractor Function
First, implement an extractor function in `vllm_omni/diffusion/cache/teacache/extractors.py`. The extractor extracts the modulated input and defines how to run transformer blocks:
```python
def extract_your_model_context(
module: nn.Module,
hidden_states: torch.Tensor,
timestep: torch.Tensor,
**kwargs: Any,
) -> CacheContext:
# 1. Preprocessing
temb = module.time_embed(timestep)
# 2. Extract modulated input (for cache decision)
modulated_input = module.transformer_blocks[0].norm1(hidden_states, temb)
# 3. Define transformer execution
def run_transformer_blocks():
h = hidden_states
for block in module.transformer_blocks:
h = block(h, temb=temb)
return (h,)
# 4. Define postprocessing
def postprocess(h):
return module.proj_out(module.norm_out(h, temb))
return CacheContext(
modulated_input=modulated_input,
hidden_states=hidden_states,
encoder_hidden_states=None,
temb=temb,
run_transformer_blocks=run_transformer_blocks,
postprocess=postprocess,
)
```
Register it in `EXTRACTOR_REGISTRY`:
```python
EXTRACTOR_REGISTRY = {
...
"YourTransformer2DModel": extract_your_model_context,
}
```
### 5.2 Add Adapter for Coefficient Estimation
Add an adapter in `vllm_omni/diffusion/cache/teacache/coefficient_estimator.py`:
```python
class YourModelAdapter:
@staticmethod
def load_pipeline(model_path: str, device: str, dtype: torch.dtype) -> Any:
# Load your pipeline
...
@staticmethod
def get_transformer(pipeline: Any) -> tuple[Any, str]:
return pipeline.transformer, "YourTransformer2DModel"
@staticmethod
def install_hook(transformer: Any, hook: DataCollectionHook) -> None:
registry = HookRegistry.get_or_create(transformer)
registry.register_hook(hook._HOOK_NAME, hook)
_MODEL_ADAPTERS["YourModel"] = YourModelAdapter
```
### 5.3 Run Coefficient Estimation
Use the provided script to estimate coefficients:
```python
from vllm_omni.diffusion.cache.teacache.coefficient_estimator import (
TeaCacheCoefficientEstimator,
)
from datasets import load_dataset
from tqdm import tqdm
# Load model
estimator = TeaCacheCoefficientEstimator(
model_path="/path/to/model",
model_type="Bagel", # Your model type
device="cuda",
)
# Load prompts (paper suggests ~70 prompts)
dataset = load_dataset("nateraw/parti-prompts", split="train")
prompts = dataset["Prompt"][:70]
# Collect data
for prompt in tqdm(prompts):
estimator.collect_from_prompt(prompt, num_inference_steps=50)
# Estimate coefficients
coeffs = estimator.estimate(poly_order=4)
print(f"Coefficients: {coeffs}")
```
### 5.4 Interpreting Coefficient Estimation Results
The estimator outputs statistics and polynomial coefficients. Here's how to interpret them:
**Example Output:**
```
Data statistics:
Count: 48
Input Diffs (x): min=1.1089e-02, max=5.2555e-02, mean=2.8435e-02
Output Diffs (y): min=2.8242e-02, max=2.9792e-01, mean=7.0312e-02
Coefficients: [1333131.29, -168644.23, 7950.51, -163.75, 1.26]
```
**What to Check:**
- **Count**: Number of timestep pairs analyzed. Should be at least 30-50 for reliable estimation. Low count suggests insufficient prompts or inference steps.
- **Input/Output Ranges**: Verify output differences correlate with input differences. If ranges seem unusual, check your prompt diversity.
- **Coefficient Magnitude**: Extremely large values (>1e8) may indicate numerical instability - try collecting more diverse data.
**Troubleshooting:**
- If results seem unreliable, try:
- Increasing number of prompts (100+ recommended)
- Using more diverse prompts from multiple datasets
- Adjusting `num_inference_steps` (try 20, 50, 100)
### 5.5 Add Coefficients to Config
Add the estimated coefficients to `vllm_omni/diffusion/cache/teacache/config.py`:
```python
_MODEL_COEFFICIENTS = {
...
"YourTransformer2DModel": [
1.04730573e+06, # a4
-1.34150749e+05, # a3
6.51517806e+03, # a2
-1.41209108e+02, # a1
1.17241808e+00, # a0
],
}
```
## Step 6: Open a Pull Request
When submitting a pull request to add support for a new model, please include the following information in the PR description:
@@ -17,7 +17,7 @@ Therefore, it is recommended to install vLLM and vLLM-Omni with a **fresh new**
# --8<-- [start:pre-built-wheels]
#### Installation of vLLM
Note: Pre-built wheels are currently only available for vLLM-Omni 0.11.0rc1, 0.12.0rc1, 0.14.0rc1. For the latest version, please [build from source](https://docs.vllm.ai/projects/vllm-omni/en/latest/getting_started/installation/gpu/#build-wheel-from-source).
Note: Pre-built wheels are currently only available for vLLM-Omni 0.11.0rc1, 0.12.0rc1, 0.14.0rc1, 0.14.0. For the latest version, please [build from source](https://docs.vllm.ai/projects/vllm-omni/en/latest/getting_started/installation/gpu/#build-wheel-from-source).
vLLM-Omni is built based on vLLM. Please install it with command below.
@@ -93,7 +93,7 @@ docker run --runtime nvidia --gpus 2 \
--env "HF_TOKEN=$HF_TOKEN" \
-p 8091:8091 \
--ipc=host \
vllm/vllm-omni:v0.14.0rc1 \
vllm/vllm-omni:v0.14.0 \
--model Qwen/Qwen3-Omni-30B-A3B-Instruct --port 8091
```
@@ -80,7 +80,7 @@ docker run --rm \
-v ~/.cache/huggingface:/root/.cache/huggingface \
--env "HF_TOKEN=$HF_TOKEN" \
-p 8091:8091 \
vllm/vllm-omni-rocm:v0.14.0rc1 \
vllm/vllm-omni-rocm:v0.14.0 \
--model Qwen/Qwen3-Omni-30B-A3B-Instruct --omni --port 8091
```
@@ -99,7 +99,7 @@ docker run --rm -it \
-v ~/.cache/huggingface:/root/.cache/huggingface \
--env "HF_TOKEN=$HF_TOKEN" \
--entrypoint bash \
vllm/vllm-omni-rocm:v0.14.0rc1
vllm/vllm-omni-rocm:v0.14.0
```
# --8<-- [end:pre-built-images]
+3 -2
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "vllm-omni"
version = "0.14.0rc1"
version = "0.14.0"
description = "A framework for efficient model inference with omni-modality models"
readme = "README.md"
requires-python = ">=3.10,<3.14"
@@ -59,7 +59,8 @@ dev = [
"psutil>=7.2.0",
"soundfile>=0.13.1",
"imageio[ffmpeg]>=0.6.0",
"opencv-python>=4.12.0.88"
"opencv-python>=4.12.0.88",
"mooncake-transfer-engine==0.3.8.post1"
]
docs = [
@@ -0,0 +1,85 @@
# stage config for running BAGEL with Mooncake connector for CI e2e tests.
# This config is optimized for single GPU tests with Mooncake inter-stage communication.
stage_args:
- stage_id: 0
stage_type: llm
runtime:
devices: "0"
max_batch_size: 1
engine_args:
model_stage: thinker
model_arch: BagelForConditionalGeneration
worker_type: ar
scheduler_cls: vllm_omni.core.sched.omni_ar_scheduler.OmniARScheduler
gpu_memory_utilization: 0.60
enforce_eager: true
trust_remote_code: true
engine_output_type: text
distributed_executor_backend: mp
enable_prefix_caching: false
max_num_batched_tokens: 32768
tensor_parallel_size: 1
omni_kv_config:
need_send_cache: true
kv_transfer_criteria:
type: prefill_finished
final_output: true
final_output_type: text
is_comprehension: true
default_sampling_params:
temperature: 0.4
top_p: 0.9
top_k: 1
max_tokens: 2048
seed: 52
detokenize: true
repetition_penalty: 1.05
output_connectors:
to_stage_1: mooncake_connector
- stage_id: 1
stage_type: diffusion
runtime:
devices: "0"
max_batch_size: 1
engine_args:
model_stage: dit
gpu_memory_utilization: 0.4
enforce_eager: true
trust_remote_code: true
engine_output_type: image
distributed_executor_backend: mp
enable_prefix_caching: false
max_num_batched_tokens: 32768
tensor_parallel_size: 1
omni_kv_config:
need_recv_cache: true
engine_input_source: [0]
final_output: true
final_output_type: image
is_comprehension: false
default_sampling_params:
seed: 52
input_connectors:
from_stage_0: mooncake_connector
# Top-level runtime config with Mooncake connector
runtime:
enabled: true
defaults:
window_size: -1
max_inflight: 1
connectors:
mooncake_connector:
name: MooncakeConnector
extra:
host: "${MOONCAKE_HOST}"
metadata_server: "http://${MOONCAKE_HOST}:${MOONCAKE_HTTP_PORT}/metadata"
master: "${MOONCAKE_HOST}:${MOONCAKE_RPC_PORT}"
segment: 64000000
localbuf: 64000000
proto: tcp
edges:
- from: 0
to: 1
window_size: -1
@@ -0,0 +1,83 @@
# stage config for running BAGEL with SharedMemory connector for CI e2e tests.
# This config is optimized for single GPU tests with SharedMemory inter-stage communication.
stage_args:
- stage_id: 0
stage_type: llm
runtime:
devices: "0"
max_batch_size: 1
engine_args:
model_stage: thinker
model_arch: BagelForConditionalGeneration
worker_type: ar
scheduler_cls: vllm_omni.core.sched.omni_ar_scheduler.OmniARScheduler
gpu_memory_utilization: 0.65
enforce_eager: true
trust_remote_code: true
engine_output_type: text
distributed_executor_backend: "mp"
enable_prefix_caching: false
max_num_batched_tokens: 32768
tensor_parallel_size: 1
omni_kv_config:
need_send_cache: true
kv_transfer_criteria:
type: prefill_finished #or special token generated
final_output: true
final_output_type: text
is_comprehension: true
default_sampling_params:
temperature: 0.4
top_p: 0.9
top_k: 1
max_tokens: 2048
seed: 52
detokenize: True
repetition_penalty: 1.05
- stage_id: 1
stage_type: diffusion
runtime:
devices: "0"
max_batch_size: 1
engine_args:
model_stage: dit
gpu_memory_utilization: 0.4
enforce_eager: true
trust_remote_code: true
engine_output_type: image
distributed_executor_backend: "mp"
enable_prefix_caching: false
max_num_batched_tokens: 32768
tensor_parallel_size: 1
omni_kv_config:
need_recv_cache: true
engine_input_source: [0]
final_output: true
final_output_type: image
is_comprehension: false
default_sampling_params:
seed: 52
# Runtime edges
runtime:
enabled: true
defaults:
window_size: -1
max_inflight: 1
# Distributed connectors configuration (optional)
# More connectors will be supported in the future.
connectors:
shared_memory_connector:
name: SharedMemoryConnector
extra:
shm_threshold_bytes: 65536 # 64KB threshold
edges:
- from: 0
to: 1
window_size: -1
@@ -0,0 +1,308 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""
End-to-end test for Bagel text2img generation.
This test validates that the Bagel model generates images that match
expected reference pixel values within a ±5 tolerance.
Equivalent to running:
python3 examples/offline_inference/bagel/end2end.py \
--prompts "A futuristic city skyline at twilight, cyberpunk style" \
--modality text2img --step 15
"""
import os
os.environ["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn"
os.environ["VLLM_TEST_CLEAN_GPU_MEMORY"] = "1"
import signal
import socket
import subprocess
import tempfile
import time
from pathlib import Path
from typing import Any
import pytest
from PIL import Image
from tests.utils import hardware_test
from vllm_omni.entrypoints.omni import Omni
# Reference pixel data extracted from the known-good output image
# Each entry contains (x, y) position and expected (R, G, B) values
# "Generated with seed=52, num_inference_steps=15,
# prompt='A futuristic city skyline at twilight, cyberpunk style'"
REFERENCE_PIXELS = [
{"position": (100, 100), "rgb": (68, 107, 134)},
{"position": (400, 50), "rgb": (95, 139, 166)},
{"position": (700, 100), "rgb": (99, 122, 151)},
{"position": (150, 400), "rgb": (111, 125, 153)},
{"position": (512, 512), "rgb": (97, 107, 131)},
{"position": (700, 400), "rgb": (48, 64, 98)},
{"position": (100, 700), "rgb": (79, 63, 84)},
{"position": (400, 700), "rgb": (40, 58, 79)},
{"position": (700, 700), "rgb": (60, 75, 103)},
{"position": (256, 256), "rgb": (97, 128, 156)},
]
# Maximum allowed difference per color channel
PIXEL_TOLERANCE = 5
# Default test prompt
DEFAULT_PROMPT = "<|im_start|>A futuristic city skyline at twilight, cyberpunk style<|im_end|>"
def _find_free_port() -> int:
"""Find and return a free ephemeral port by binding to port 0."""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(("127.0.0.1", 0))
s.listen(1)
port = s.getsockname()[1]
return port
def _configure_sampling_params(omni: Omni, max_tokens: int = 1, num_inference_steps: int = 15) -> list:
"""Configure sampling parameters for Bagel text2img generation.
Args:
omni: The Omni instance to get default params from.
max_tokens: Maximum tokens for the first stage.
num_inference_steps: Number of inference steps for the diffusion stage.
Returns:
Configured sampling params list.
"""
params_list = omni.default_sampling_params_list
params_list[0].max_tokens = max_tokens # type: ignore
if len(params_list) > 1:
params_list[1].num_inference_steps = num_inference_steps # type: ignore
return params_list
def _extract_generated_image(omni_outputs: list) -> Image.Image | None:
"""Extract the generated image from Omni outputs.
Args:
omni_outputs: List of outputs from omni.generate().
Returns:
The first generated PIL Image, or None if no image found.
"""
for req_output in omni_outputs:
if images := getattr(req_output, "images", None):
return images[0]
if hasattr(req_output, "request_output") and req_output.request_output:
for stage_out in req_output.request_output:
if hasattr(stage_out, "images") and stage_out.images:
return stage_out.images[0]
return None
def _validate_pixels(
image: Image.Image,
reference_pixels: list[dict[str, Any]] = REFERENCE_PIXELS,
tolerance: int = PIXEL_TOLERANCE,
) -> None:
"""Validate that image pixels match expected reference values.
Args:
image: The PIL Image to validate.
reference_pixels: List of dicts with 'position' (x, y) and 'rgb' (R, G, B).
tolerance: Maximum allowed difference per color channel.
Raises:
AssertionError: If any pixel differs beyond tolerance.
"""
for ref in reference_pixels:
x, y = ref["position"]
expected = ref["rgb"]
actual = image.getpixel((x, y))[:3]
assert all(abs(a - e) <= tolerance for a, e in zip(actual, expected)), (
f"Pixel mismatch at ({x}, {y}): expected {expected}, got {actual}"
)
def _generate_bagel_image(omni: Omni, prompt: str = DEFAULT_PROMPT) -> Image.Image:
"""Generate an image using Bagel model with configured parameters.
Args:
omni: The Omni instance to use for generation.
prompt: The text prompt for image generation.
Returns:
The generated PIL Image.
Raises:
AssertionError: If no image is generated or size is incorrect.
"""
params_list = _configure_sampling_params(omni)
omni_outputs = list(
omni.generate(
prompts=[{"prompt": prompt, "modalities": ["image"]}],
sampling_params_list=params_list,
)
)
generated_image = _extract_generated_image(omni_outputs)
assert generated_image is not None, "No images generated"
assert generated_image.size == (1024, 1024), f"Expected 1024x1024, got {generated_image.size}"
return generated_image
@pytest.mark.core_model
@pytest.mark.diffusion
@hardware_test(res={"cuda": "H100"})
def test_bagel_text2img_shared_memory_connector():
"""Test Bagel text2img with shared memory connector."""
config_path = str(Path(__file__).parent / "stage_configs" / "bagel_sharedmemory_ci.yaml")
omni = Omni(model="ByteDance-Seed/BAGEL-7B-MoT", stage_configs_path=config_path)
try:
generated_image = _generate_bagel_image(omni)
_validate_pixels(generated_image)
finally:
omni.close()
def _wait_for_port(host: str, port: int, timeout: int = 30) -> bool:
"""Wait for a port to become available.
Args:
host: The host address.
port: The port number.
timeout: Maximum seconds to wait.
Returns:
True if port becomes available, False otherwise.
"""
for _ in range(timeout):
try:
with socket.create_connection((host, port), timeout=1):
return True
except (TimeoutError, ConnectionRefusedError):
time.sleep(1)
return False
def _cleanup_mooncake_processes(timeout_secs: int = 5) -> None:
"""Clean up any existing mooncake_master processes.
Args:
timeout_secs: Maximum seconds to wait for graceful termination.
"""
subprocess.run(
["pkill", "-f", "mooncake_master"],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
start_time = time.time()
while time.time() - start_time < timeout_secs:
result = subprocess.run(
["pgrep", "-f", "mooncake_master"],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
if result.returncode != 0:
break
time.sleep(0.5)
else:
subprocess.run(
["pkill", "-9", "-f", "mooncake_master"],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
time.sleep(1)
def _load_mooncake_config(host: str, rpc_port: int, http_port: int) -> str:
"""Load Mooncake config from YAML and substitute placeholders.
Args:
host: Mooncake host address.
rpc_port: RPC port for Mooncake master.
http_port: HTTP metadata server port.
Returns:
Path to the temporary config file with substituted values.
"""
config_path = str(Path(__file__).parent / "stage_configs" / "bagel_mooncake_ci.yaml")
with open(config_path) as f:
config_content = f.read()
# Substitute placeholders
config_content = config_content.replace("${MOONCAKE_HOST}", host)
config_content = config_content.replace("${MOONCAKE_RPC_PORT}", str(rpc_port))
config_content = config_content.replace("${MOONCAKE_HTTP_PORT}", str(http_port))
# Write to temp file
temp_file = tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False)
temp_file.write(config_content)
temp_file.close()
return temp_file.name
@pytest.mark.core_model
@pytest.mark.diffusion
@hardware_test(res={"cuda": "H100"})
def test_bagel_text2img_mooncake_connector():
"""Test Bagel text2img with Mooncake connector for inter-stage communication."""
MOONCAKE_HOST = "127.0.0.1"
MOONCAKE_RPC_PORT = _find_free_port()
MOONCAKE_HTTP_PORT = _find_free_port()
MOONCAKE_METRICS_PORT = _find_free_port()
mooncake_master_proc = None
temp_config_file = None
omni = None
try:
_cleanup_mooncake_processes()
# Start mooncake_master
mooncake_master_proc = subprocess.Popen(
[
"mooncake_master",
f"--rpc_port={MOONCAKE_RPC_PORT}",
"--enable_http_metadata_server=true",
"--http_metadata_server_host=0.0.0.0",
f"--http_metadata_server_port={MOONCAKE_HTTP_PORT}",
f"--metrics_port={MOONCAKE_METRICS_PORT}",
],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
preexec_fn=os.setsid,
)
assert _wait_for_port(MOONCAKE_HOST, MOONCAKE_RPC_PORT), "mooncake_master failed to start"
# Create temp config and initialize Omni
temp_config_file = _load_mooncake_config(
host=MOONCAKE_HOST,
rpc_port=MOONCAKE_RPC_PORT,
http_port=MOONCAKE_HTTP_PORT,
)
omni = Omni(model="ByteDance-Seed/BAGEL-7B-MoT", stage_configs_path=temp_config_file)
generated_image = _generate_bagel_image(omni)
_validate_pixels(generated_image)
finally:
if omni:
omni.close()
if temp_config_file:
try:
os.unlink(temp_config_file)
except OSError:
pass
if mooncake_master_proc:
try:
os.killpg(os.getpgid(mooncake_master_proc.pid), signal.SIGKILL)
except OSError:
pass
@@ -0,0 +1,197 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
import types
from typing import Any
import numpy as np
import torch
from vllm.config import LoadConfig
from vllm_omni.diffusion.cache.teacache.extractors import get_extractor
from vllm_omni.diffusion.data import OmniDiffusionConfig
from vllm_omni.diffusion.hooks import HookRegistry, ModelHook
from vllm_omni.diffusion.model_loader.diffusers_loader import DiffusersPipelineLoader
from vllm_omni.diffusion.models.bagel.pipeline_bagel import BagelPipeline
class DataCollectionHook(ModelHook):
"""Hook to collect modulated inputs and model outputs for TeaCache coefficient estimation."""
_HOOK_NAME = "teacache_collector"
def __init__(self, transformer_type: str):
super().__init__()
self.transformer_type = transformer_type
self.extractor_fn = None
self.current_trajectory: list[tuple[np.ndarray, np.ndarray]] = []
def initialize_hook(self, module: torch.nn.Module) -> torch.nn.Module:
self.extractor_fn = get_extractor(self.transformer_type)
return module
def new_forward(self, module: torch.nn.Module, *args: Any, **kwargs: Any) -> Any:
ctx = self.extractor_fn(module, *args, **kwargs)
modulated_input_cpu = ctx.modulated_input.detach().cpu().numpy()
outputs = ctx.run_transformer_blocks()
ctx.hidden_states = outputs[0]
if len(outputs) > 1 and ctx.encoder_hidden_states is not None:
ctx.encoder_hidden_states = outputs[1]
model_output_cpu = ctx.hidden_states.detach().cpu().numpy()
self.current_trajectory.append((modulated_input_cpu, model_output_cpu))
return ctx.postprocess(ctx.hidden_states)
def start_collection(self):
self.current_trajectory = []
def stop_collection(self) -> list[tuple[np.ndarray, np.ndarray]]:
return list(self.current_trajectory)
class BagelAdapter:
"""Adapter for Bagel model."""
@staticmethod
def load_pipeline(model_path: str, device: str = "cuda", dtype: torch.dtype = torch.bfloat16) -> BagelPipeline:
od_config = OmniDiffusionConfig.from_kwargs(model=model_path, dtype=dtype)
od_config.model_class_name = "BagelPipeline"
pipeline = BagelPipeline(od_config=od_config)
loader = DiffusersPipelineLoader(LoadConfig())
loader.load_weights(pipeline)
pipeline.to(device)
return pipeline
@staticmethod
def get_transformer(pipeline: Any) -> tuple[Any, str]:
return pipeline.bagel, "Bagel"
@staticmethod
def install_hook(transformer: Any, hook: DataCollectionHook) -> None:
original_forward_flow = transformer._forward_flow
def forward_alias(self, *args, **kwargs):
return original_forward_flow(*args, **kwargs)
transformer.forward = types.MethodType(forward_alias, transformer)
registry = HookRegistry.get_or_create(transformer)
registry.register_hook(hook._HOOK_NAME, hook)
transformer._forward_flow = transformer.forward
class DefaultAdapter:
"""Default adapter for standard diffusers pipelines."""
@staticmethod
def load_pipeline(model_path: str, device: str, dtype: torch.dtype) -> Any:
raise NotImplementedError("DefaultAdapter.load_pipeline not implemented")
@staticmethod
def get_transformer(pipeline: Any) -> tuple[Any, str]:
return pipeline.transformer, pipeline.transformer.__class__.__name__
@staticmethod
def install_hook(transformer: Any, hook: DataCollectionHook) -> None:
registry = HookRegistry.get_or_create(transformer)
registry.register_hook(hook._HOOK_NAME, hook)
_MODEL_ADAPTERS: dict[str, type] = {
"Bagel": BagelAdapter,
}
_EPSILON = 1e-6
def calculate_relative_l1(tensor_current: np.ndarray, tensor_next: np.ndarray) -> float:
"""Calculate relative L1 distance (Eq. 4 from TeaCache paper)."""
diff = np.abs(tensor_current - tensor_next).sum()
norm = np.abs(tensor_current).sum() + _EPSILON
return diff / norm
def estimate_teacache_coefficients(
collected_data: list[list[tuple[np.ndarray, np.ndarray]]], poly_order: int = 4
) -> list[float]:
"""Estimate polynomial coefficients for TeaCache using np.polyfit."""
input_diffs, output_diffs = [], []
for sample in collected_data:
for t in range(len(sample) - 1):
feat_in_curr, feat_out_curr = sample[t]
feat_in_next, feat_out_next = sample[t + 1]
input_diffs.append(calculate_relative_l1(feat_in_curr, feat_in_next))
output_diffs.append(calculate_relative_l1(feat_out_curr, feat_out_next))
x = np.array(input_diffs, dtype=np.float64)
y = np.array(output_diffs, dtype=np.float64)
print("Data statistics:")
print(f" Count: {len(x)}")
print(f" Input Diffs (x): min={x.min():.4e}, max={x.max():.4e}, mean={x.mean():.4e}")
print(f" Output Diffs (y): min={y.min():.4e}, max={y.max():.4e}, mean={y.mean():.4e}")
return np.polyfit(x, y, poly_order).tolist()
class TeaCacheCoefficientEstimator:
"""Model-agnostic helper class to collect data and estimate TeaCache coefficients."""
def __init__(
self,
model_path: str,
model_type: str = "Bagel",
device: str = "cuda",
dtype: torch.dtype = torch.bfloat16,
):
# Add validation here ⬇️
if model_type not in _MODEL_ADAPTERS:
available_types = list(_MODEL_ADAPTERS.keys())
raise ValueError(
f"Unsupported model_type: '{model_type}'. "
f"Available types: {available_types}. "
f"To add support for a new model, add an entry to _MODEL_ADAPTERS."
)
adapter = _MODEL_ADAPTERS.get(model_type, DefaultAdapter)
self.pipeline = adapter.load_pipeline(model_path, device, dtype)
self.transformer, self.transformer_type = adapter.get_transformer(self.pipeline)
self.hook = DataCollectionHook(self.transformer_type)
self.collected_data: list[list[tuple[np.ndarray, np.ndarray]]] = []
adapter.install_hook(self.transformer, self.hook)
def collect_from_prompt(self, prompt: str, **generate_kwargs):
self.hook.start_collection()
from vllm_omni.diffusion.request import OmniDiffusionRequest
req = OmniDiffusionRequest(
prompt=prompt,
num_inference_steps=generate_kwargs.get("num_inference_steps", 20),
seed=generate_kwargs.get("seed", 42),
)
self.pipeline.forward(req)
trajectory = self.hook.stop_collection()
if trajectory:
self.collected_data.append(trajectory)
def estimate(self, poly_order: int = 4) -> list[float]:
"""Estimate polynomial coefficients from collected data.
Args:
poly_order: Order of polynomial fit (default: 4)
Returns:
List of polynomial coefficients [a_n, a_{n-1}, ..., a_1, a_0]
Raises:
RuntimeError: If no data has been collected
"""
if not self.collected_data:
raise RuntimeError(
"No data collected for coefficient estimation. "
"Call collect_from_prompt() at least once before calling estimate()."
)
return estimate_teacache_coefficients(self.collected_data, poly_order)
@@ -11,7 +11,7 @@ stage_args:
model_arch: BagelForConditionalGeneration
worker_type: ar
scheduler_cls: vllm_omni.core.sched.omni_ar_scheduler.OmniARScheduler
gpu_memory_utilization: 0.4
gpu_memory_utilization: 0.35
enforce_eager: true
trust_remote_code: true
engine_output_type: text
@@ -42,7 +42,7 @@ stage_args:
max_batch_size: 1
engine_args:
model_stage: dit
gpu_memory_utilization: 0.4
gpu_memory_utilization: 0.55
enforce_eager: true
trust_remote_code: true
engine_output_type: image
@@ -11,7 +11,7 @@ stage_args:
model_arch: BagelForConditionalGeneration
worker_type: ar
scheduler_cls: vllm_omni.core.sched.omni_ar_scheduler.OmniARScheduler
gpu_memory_utilization: 0.4
gpu_memory_utilization: 0.35
enforce_eager: true
trust_remote_code: true
engine_output_type: text
@@ -41,11 +41,11 @@ stage_args:
- stage_id: 1
stage_type: diffusion
runtime:
devices: "1"
devices: "0"
max_batch_size: 1
engine_args:
model_stage: dit
gpu_memory_utilization: 0.4
gpu_memory_utilization: 0.55
enforce_eager: true
trust_remote_code: true
engine_output_type: image
+2 -2
View File
@@ -1,3 +1,3 @@
__version__ = "0.14.0rc1"
__version_tuple__ = (0, 14, "0rc1")
__version__ = "0.14.0"
__version_tuple__ = (0, 14, 0)
# TODO: add auto version generation