[Feat][Benchmark] Add synchronous video generation endpoint POST /v1/videos/sync for benchmark test (#2049)

Signed-off-by: samithuang <285365963@qq.com>
This commit is contained in:
Samit authored and GitHub committed 2026-04-01 10:47:33 +08:00
1 parent 7274e15840
commit 7b965a7b6c
7 files changed
+421 -62

No files matched your search

@@ -37,12 +37,45 @@ artifact, poll the job status and then download the completed file from the
content endpoint.
The main endpoints are:
- `POST /v1/videos`: create a video generation job
- `POST /v1/videos`: create a video generation job (async)
- `POST /v1/videos/sync`: generate a video and return raw bytes (sync, for benchmarks)
- `GET /v1/videos/{video_id}`: retrieve the current job status and metadata
- `GET /v1/videos`: list stored video jobs
- `GET /v1/videos/{video_id}/content`: download the generated video file
- `DELETE /v1/videos/{video_id}`: delete the job and any stored output
## Sync API (Benchmark / Testing)
`POST /v1/videos/sync` is a synchronous alternative that blocks until generation
completes and returns the raw video bytes (`video/mp4`) directly in the response
body. It is designed for benchmark and testing scenarios where one-shot
request/response latency measurement is needed.
The sync endpoint accepts the same form parameters as `POST /v1/videos`. It does
not create any stored job record — the response is purely the generated video
file. Metadata is returned via response headers:
- `X-Request-Id`: unique identifier for this generation request
- `X-Model`: model name used for generation
- `X-Inference-Time-S`: wall-clock inference time in seconds
```bash
curl -X POST http://localhost:8091/v1/videos/sync \
-F "prompt=A bear playing with yarn, smooth motion" \
-F "input_reference=@/path/to/input.png" \
-F "size=832x480" \
-F "num_frames=33" \
-F "fps=16" \
-F "negative_prompt=low quality, blurry, static" \
-F "num_inference_steps=40" \
-F "guidance_scale=1.0" \
-F "guidance_scale_2=1.0" \
-F "boundary_ratio=0.875" \
-F "flow_shift=12.0" \
-F "seed=42" \
-o sync_i2v_output.mp4
```
## Storage
Generated video files are stored on local disk by the async video API.
@@ -37,12 +37,43 @@ artifact, poll the job status and then download the completed file from the
content endpoint.
The main endpoints are:
- `POST /v1/videos`: create a video generation job
- `POST /v1/videos`: create a video generation job (async)
- `POST /v1/videos/sync`: generate a video and return raw bytes (sync, for benchmarks)
- `GET /v1/videos/{video_id}`: retrieve the current job status and metadata
- `GET /v1/videos`: list stored video jobs
- `GET /v1/videos/{video_id}/content`: download the generated video file
- `DELETE /v1/videos/{video_id}`: delete the job and any stored output
## Sync API (Benchmark / Testing)
`POST /v1/videos/sync` is a synchronous alternative that blocks until generation
completes and returns the raw video bytes (`video/mp4`) directly in the response
body. It is designed for benchmark and testing scenarios where one-shot
request/response latency measurement is needed.
The sync endpoint accepts the same form parameters as `POST /v1/videos`. It does
not create any stored job record — the response is purely the generated video
file. Metadata is returned via response headers:
- `X-Request-Id`: unique identifier for this generation request
- `X-Model`: model name used for generation
- `X-Inference-Time-S`: wall-clock inference time in seconds
```bash
curl -X POST http://localhost:8091/v1/videos/sync \
-F "prompt=Two anthropomorphic cats in comfy boxing gear and bright gloves fight intensely on a spotlighted stage." \
-F "size=832x480" \
-F "num_frames=33" \
-F "fps=16" \
-F "num_inference_steps=40" \
-F "guidance_scale=4.0" \
-F "guidance_scale_2=4.0" \
-F "boundary_ratio=0.875" \
-F "flow_shift=5.0" \
-F "seed=42" \
-o sync_t2v_output.mp4
```
## Storage
Generated video files are stored on local disk by the async video API.
@@ -34,12 +34,45 @@ artifact, poll the job status and then download the completed file from the
content endpoint.
The main endpoints are:
- `POST /v1/videos`: create a video generation job
- `POST /v1/videos`: create a video generation job (async)
- `POST /v1/videos/sync`: generate a video and return raw bytes (sync, for benchmarks)
- `GET /v1/videos/{video_id}`: retrieve the current job status and metadata
- `GET /v1/videos`: list stored video jobs
- `GET /v1/videos/{video_id}/content`: download the generated video file
- `DELETE /v1/videos/{video_id}`: delete the job and any stored output
## Sync API (Benchmark / Testing)
`POST /v1/videos/sync` is a synchronous alternative that blocks until generation
completes and returns the raw video bytes (`video/mp4`) directly in the response
body. It is designed for benchmark and testing scenarios where one-shot
request/response latency measurement is needed.
The sync endpoint accepts the same form parameters as `POST /v1/videos`. It does
not create any stored job record — the response is purely the generated video
file. Metadata is returned via response headers:
- `X-Request-Id`: unique identifier for this generation request
- `X-Model`: model name used for generation
- `X-Inference-Time-S`: wall-clock inference time in seconds
```bash
curl -X POST http://localhost:8091/v1/videos/sync \
-F "prompt=A bear playing with yarn, smooth motion" \
-F "input_reference=@/path/to/input.png" \
-F "size=832x480" \
-F "num_frames=33" \
-F "fps=16" \
-F "negative_prompt=low quality, blurry, static" \
-F "num_inference_steps=40" \
-F "guidance_scale=1.0" \
-F "guidance_scale_2=1.0" \
-F "boundary_ratio=0.875" \
-F "flow_shift=12.0" \
-F "seed=42" \
-o sync_i2v_output.mp4
```
## Storage
Generated video files are stored on local disk by the async video API.
@@ -34,12 +34,43 @@ artifact, poll the job status and then download the completed file from the
content endpoint.
The main endpoints are:
- `POST /v1/videos`: create a video generation job
- `POST /v1/videos`: create a video generation job (async)
- `POST /v1/videos/sync`: generate a video and return raw bytes (sync, for benchmarks)
- `GET /v1/videos/{video_id}`: retrieve the current job status and metadata
- `GET /v1/videos`: list stored video jobs
- `GET /v1/videos/{video_id}/content`: download the generated video file
- `DELETE /v1/videos/{video_id}`: delete the job and any stored output
## Sync API (Benchmark / Testing)
`POST /v1/videos/sync` is a synchronous alternative that blocks until generation
completes and returns the raw video bytes (`video/mp4`) directly in the response
body. It is designed for benchmark and testing scenarios where one-shot
request/response latency measurement is needed.
The sync endpoint accepts the same form parameters as `POST /v1/videos`. It does
not create any stored job record — the response is purely the generated video
file. Metadata is returned via response headers:
- `X-Request-Id`: unique identifier for this generation request
- `X-Model`: model name used for generation
- `X-Inference-Time-S`: wall-clock inference time in seconds
```bash
curl -X POST http://localhost:8091/v1/videos/sync \
-F "prompt=Two anthropomorphic cats in comfy boxing gear and bright gloves fight intensely on a spotlighted stage." \
-F "size=832x480" \
-F "num_frames=33" \
-F "fps=16" \
-F "num_inference_steps=40" \
-F "guidance_scale=4.0" \
-F "guidance_scale_2=4.0" \
-F "boundary_ratio=0.875" \
-F "flow_shift=5.0" \
-F "seed=42" \
-o sync_t2v_output.mp4
```
## Storage
Generated video files are stored on local disk by the async video API.
@@ -735,3 +735,156 @@ def test_extra_params_merged_with_existing_extra_args(test_client, mocker: Mocke
assert captured.extra_args["flow_shift"] == 0.5
assert captured.extra_args["use_zero_init"] is True
assert captured.extra_args["zero_steps"] == 2
# ---------------------------------------------------------------------------
# Sync endpoint tests (POST /v1/videos/sync)
# ---------------------------------------------------------------------------
def _mock_encode_video_bytes(mocker: MockerFixture, return_value: bytes = b"fake-video-bytes"):
"""Mock the raw-bytes encoder used by the sync video path."""
return mocker.patch(
"vllm_omni.entrypoints.openai.serving_video._encode_video_bytes",
return_value=return_value,
)
def test_sync_t2v_returns_video_bytes(test_client, mocker: MockerFixture):
"""Sync endpoint should block until generation finishes and return raw
video bytes with metadata headers."""
_mock_encode_video_bytes(mocker, b"fake-video-bytes")
response = test_client.post(
"/v1/videos/sync",
data={
"prompt": "A cat running across the street.",
"size": "640x360",
"seconds": "2",
"fps": "12",
},
)
assert response.status_code == 200
assert response.headers["content-type"] == "video/mp4"
assert response.content == b"fake-video-bytes"
assert response.headers["x-request-id"].startswith("video_sync-")
assert response.headers["x-model"] == "Wan-AI/Wan2.2-T2V-A14B-Diffusers"
assert float(response.headers["x-inference-time-s"]) >= 0
def test_sync_i2v_returns_video_bytes(test_client, mocker: MockerFixture):
"""Sync I2V endpoint should accept an uploaded reference image and return
raw video bytes."""
image_bytes = _make_test_image_bytes((48, 32))
_mock_encode_video_bytes(mocker, b"i2v-video-data")
response = test_client.post(
"/v1/videos/sync",
data={"prompt": "A bear playing with yarn."},
files={"input_reference": ("input.png", image_bytes, "image/png")},
)
assert response.status_code == 200
assert response.content == b"i2v-video-data"
assert response.headers["content-type"] == "video/mp4"
def test_sync_i2v_with_image_reference(test_client, mocker: MockerFixture):
"""Sync I2V endpoint should accept a JSON image_reference field."""
_mock_encode_video_bytes(mocker, b"ref-video")
response = test_client.post(
"/v1/videos/sync",
data={
"prompt": "A fox running through snow.",
"image_reference": json.dumps({"image_url": _make_test_image_data_url((40, 24))}),
},
)
assert response.status_code == 200
assert response.content == b"ref-video"
def test_sync_missing_handler_returns_503():
app = FastAPI()
app.include_router(router)
app.state.openai_serving_video = None
client = TestClient(app)
response = client.post(
"/v1/videos/sync",
data={"prompt": "no handler"},
)
assert response.status_code == 503
assert "not initialized" in response.json()["detail"].lower()
def test_sync_missing_prompt_returns_422(test_client):
response = test_client.post(
"/v1/videos/sync",
data={"size": "320x240"},
)
assert response.status_code == 422
def test_sync_rejects_both_references(test_client):
response = test_client.post(
"/v1/videos/sync",
data={
"prompt": "bad refs",
"image_reference": '{"image_url": "https://example.com/cat.png"}',
},
files={"input_reference": ("input.png", _make_test_image_bytes(), "image/png")},
)
assert response.status_code == 400
assert "either input_reference or image_reference" in response.json()["detail"].lower()
def test_sync_generation_error_returns_500(test_client, mocker: MockerFixture):
"""If the underlying generation raises, the sync endpoint should return 500."""
mocker.patch.object(
OmniOpenAIServingVideo,
"generate_video_bytes",
side_effect=RuntimeError("GPU exploded"),
)
response = test_client.post(
"/v1/videos/sync",
data={"prompt": "will fail"},
)
assert response.status_code == 500
assert "GPU exploded" in response.json()["detail"]
def test_sync_does_not_create_store_entry(test_client, mocker: MockerFixture):
"""The sync endpoint should NOT leave any record in VIDEO_STORE — it is
stateless by design."""
_mock_encode_video_bytes(mocker)
response = test_client.post(
"/v1/videos/sync",
data={"prompt": "stateless test"},
)
assert response.status_code == 200
loop = asyncio.new_event_loop()
try:
stored = loop.run_until_complete(api_server.VIDEO_STORE.list_values())
finally:
loop.close()
assert len(stored) == 0
def test_sync_sampling_params_pass_through(test_client, mocker: MockerFixture):
"""Sampling parameters should propagate to the engine through the sync path."""
_mock_encode_video_bytes(mocker)
response = test_client.post(
"/v1/videos/sync",
data={
"prompt": "param pass",
"num_inference_steps": "30",
"guidance_scale": "6.5",
"seed": "42",
},
)
assert response.status_code == 200
engine = test_client.app.state.openai_serving_video._engine_client
captured = engine.captured_sampling_params_list[0]
assert captured.num_inference_steps == 30
assert captured.guidance_scale == 6.5
assert captured.seed == 42
+90 -54
View File
@@ -20,7 +20,7 @@ from typing import Annotated, Any, Literal, cast
import httpx
import vllm.envs as envs
from fastapi import APIRouter, Depends, File, Form, HTTPException, Query, Request, UploadFile, WebSocket
from fastapi.responses import FileResponse, JSONResponse, StreamingResponse
from fastapi.responses import FileResponse, JSONResponse, Response, StreamingResponse
from PIL import Image
from pydantic import BaseModel, Field
from starlette.datastructures import State
@@ -1956,16 +1956,10 @@ async def _run_video_generation_job(
raise
@router.post(
"/v1/videos",
responses={
HTTPStatus.OK.value: {"model": VideoResponse},
HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
HTTPStatus.SERVICE_UNAVAILABLE.value: {"model": ErrorResponse},
HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
},
)
async def create_video(
VIDEO_SYNC_TIMEOUT_S = 600.0
async def _parse_video_form(
raw_request: Request,
prompt: str = Form(...),
input_reference: UploadFile | None = File(default=None),
@@ -1988,48 +1982,15 @@ async def create_video(
negative_prompt: str | None = Form(default=None),
lora: str | None = Form(default=None),
extra_params: str | None = Form(default=None),
) -> VideoResponse:
"""Create an asynchronous video generation job.
) -> tuple[VideoGenerationRequest, "OmniOpenAIServingVideo", str, ReferenceImage | None]:
"""FastAPI dependency that parses video form data, validates inputs,
resolves the handler, and decodes any reference image.
This OpenAI-style endpoint accepts multipart form-data, validates the
request payload, persists a queued job record, and starts generation in the
background. The response contains metadata for polling job status rather
than the generated video bytes.
Args:
raw_request: Raw FastAPI request for accessing app state.
prompt: Text prompt describing the requested video.
input_reference: Optional uploaded reference image file.
image_reference: Optional JSON-encoded reference image descriptor.
model: Optional model name supplied by the client.
seconds: Optional target duration string accepted by the video API.
size: Optional output size string such as ``1280x720``.
user: Optional user identifier forwarded in the stored request.
width: Optional explicit output width override.
height: Optional explicit output height override.
num_frames: Optional explicit frame count override.
fps: Optional explicit frame rate override.
num_inference_steps: Optional inference step override.
guidance_scale: Optional primary guidance scale override.
guidance_scale_2: Optional secondary guidance scale override.
boundary_ratio: Optional boundary ratio override.
flow_shift: Optional flow shift override.
true_cfg_scale: Optional true CFG scale override.
seed: Optional random seed override.
negative_prompt: Optional negative prompt.
lora: Optional JSON-encoded per-request LoRA configuration.
extra_params: Optional model-specific parameters passed directly to the model's extra_args.
Returns:
A queued ``VideoResponse`` that includes the generated job identifier
and initial metadata for later retrieval.
Raises:
HTTPException: If the request is invalid, the video handler is
unavailable, or job initialization fails.
Used by both ``POST /v1/videos`` (async) and ``POST /v1/videos/sync``.
"""
input_reference_bytes = await input_reference.read() if input_reference is not None else None
parsed_image_reference = _parse_form_json(image_reference)
if parsed_image_reference is not None and input_reference_bytes is not None:
raise HTTPException(
status_code=HTTPStatus.BAD_REQUEST.value,
@@ -2058,7 +2019,6 @@ async def create_video(
"lora": _parse_form_json(lora, expected_type=dict),
"extra_params": _parse_form_json(extra_params, expected_type=dict),
}
request_data = {k: v for k, v in request_data.items() if v is not None}
request = VideoGenerationRequest(**request_data)
@@ -2082,25 +2042,101 @@ async def create_video(
except HTTPException:
raise
except Exception as e:
logger.exception("Video generation failed: %s", e)
logger.exception("Video generation setup failed: %s", e)
raise HTTPException(
status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
detail=f"Video generation failed: {str(e)}",
detail=f"Video generation setup failed: {str(e)}",
)
ref = video_response_from_request(effective_model_name, request)
try:
image_data = await decode_input_reference(request.image_reference, input_reference_bytes)
except InvalidInputReferenceError as exc:
raise HTTPException(400, detail=str(exc) or "Invalid input reference.") from exc
reference_image = ReferenceImage(data=image_data) if image_data is not None else image_data
reference_image = ReferenceImage(data=image_data) if image_data is not None else None
return request, handler, effective_model_name, reference_image
@router.post(
"/v1/videos",
responses={
HTTPStatus.OK.value: {"model": VideoResponse},
HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
HTTPStatus.SERVICE_UNAVAILABLE.value: {"model": ErrorResponse},
HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
},
)
async def create_video(
ctx: tuple[VideoGenerationRequest, OmniOpenAIServingVideo, str, ReferenceImage | None] = Depends(_parse_video_form),
) -> VideoResponse:
"""Create an asynchronous video generation job.
Accepts multipart form-data (see ``_parse_video_form`` for parameters),
persists a queued job record, and starts generation in the background.
"""
request, handler, effective_model_name, reference_image = ctx
ref = video_response_from_request(effective_model_name, request)
await VIDEO_STORE.upsert(ref.id, ref)
task = asyncio.create_task(_run_video_generation_job(handler, request, ref.id, reference_image))
await VIDEO_TASKS.upsert(ref.id, task)
return ref
@router.post(
"/v1/videos/sync",
responses={
HTTPStatus.OK.value: {"content": {"video/mp4": {}}},
HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
HTTPStatus.SERVICE_UNAVAILABLE.value: {"model": ErrorResponse},
HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
},
)
async def create_video_sync(
ctx: tuple[VideoGenerationRequest, OmniOpenAIServingVideo, str, ReferenceImage | None] = Depends(_parse_video_form),
) -> Response:
"""Synchronous video generation endpoint.
Accepts the same form parameters as ``POST /v1/videos`` but blocks until
generation completes and returns raw video bytes (``video/mp4``) directly.
Designed for benchmark and testing scenarios.
Metadata is returned via response headers ``X-Request-Id``,
``X-Model``, and ``X-Inference-Time-S``.
"""
request, handler, effective_model_name, reference_image = ctx
request_id = f"video_sync-{random_uuid()}"
started_at = time.perf_counter()
try:
video_bytes = await asyncio.wait_for(
handler.generate_video_bytes(request, request_id, reference_image=reference_image),
timeout=VIDEO_SYNC_TIMEOUT_S,
)
except asyncio.TimeoutError:
raise HTTPException(
status_code=HTTPStatus.GATEWAY_TIMEOUT.value,
detail=f"Video generation timed out after {VIDEO_SYNC_TIMEOUT_S}s.",
)
except HTTPException:
raise
except Exception as exc:
logger.exception("Sync video generation failed for request_id=%s", request_id)
raise HTTPException(
status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value,
detail=f"Video generation failed: {str(exc)}",
) from exc
inference_time_s = time.perf_counter() - started_at
return Response(
content=video_bytes,
media_type="video/mp4",
headers={
"X-Request-Id": request_id,
"X-Model": effective_model_name,
"X-Inference-Time-S": f"{inference_time_s:.3f}",
},
)
@router.get("/v1/videos", response_model=VideoListResponse)
async def list_videos(
after: str | None = None,
+46 -4
View File
@@ -20,7 +20,7 @@ from vllm_omni.entrypoints.openai.protocol.videos import (
VideoGenerationResponse,
)
from vllm_omni.entrypoints.openai.utils import get_stage_type, parse_lora_request
from vllm_omni.entrypoints.openai.video_api_utils import encode_video_base64
from vllm_omni.entrypoints.openai.video_api_utils import _encode_video_bytes, encode_video_base64
from vllm_omni.inputs.data import OmniDiffusionSamplingParams, OmniSamplingParams, OmniTextPrompt
logger = init_logger(__name__)
@@ -71,13 +71,18 @@ class OmniOpenAIServingVideo:
stage_configs=stage_configs,
)
async def generate_videos(
async def _run_and_extract(
self,
request: VideoGenerationRequest,
reference_id: str,
*,
reference_image: ReferenceImage | None = None,
) -> VideoGenerationResponse:
) -> tuple[list[Any], list[Any | None], int, int]:
"""Run the generation pipeline and extract video/audio outputs.
Returns:
Tuple of (videos, audios, audio_sample_rate, output_fps).
"""
prompt: OmniTextPrompt = OmniTextPrompt(prompt=request.prompt)
if request.negative_prompt is not None:
prompt["negative_prompt"] = request.negative_prompt
@@ -144,12 +149,23 @@ class OmniOpenAIServingVideo:
)
result = await self._run_generation(prompt, gen_params, reference_id)
_t_encode_start = time.perf_counter()
videos = self._extract_video_outputs(result)
audios = self._extract_audio_outputs(result, expected_count=len(videos))
audio_sample_rate = self._resolve_audio_sample_rate(result)
output_fps = vp.fps or 24
return videos, audios, audio_sample_rate, output_fps
async def generate_videos(
self,
request: VideoGenerationRequest,
reference_id: str,
*,
reference_image: ReferenceImage | None = None,
) -> VideoGenerationResponse:
videos, audios, audio_sample_rate, output_fps = await self._run_and_extract(
request, reference_id, reference_image=reference_image
)
_t_encode_start = time.perf_counter()
video_data = [
VideoData(
b64_json=(
@@ -169,6 +185,32 @@ class OmniOpenAIServingVideo:
logger.info("Video response encoding (MP4+base64): %.2f ms", _t_encode_ms)
return VideoGenerationResponse(created=int(time.time()), data=video_data)
async def generate_video_bytes(
self,
request: VideoGenerationRequest,
reference_id: str,
*,
reference_image: ReferenceImage | None = None,
) -> bytes:
"""Generate a video and return raw MP4 bytes, bypassing base64 encoding."""
videos, audios, audio_sample_rate, output_fps = await self._run_and_extract(
request, reference_id, reference_image=reference_image
)
if len(videos) > 1:
logger.warning(
"Video request %s generated %d outputs; returning only the first.", reference_id, len(videos)
)
audio = audios[0]
_t_encode_start = time.perf_counter()
video_bytes = _encode_video_bytes(
videos[0],
fps=output_fps,
**({"audio": audio, "audio_sample_rate": audio_sample_rate} if audio is not None else {}),
)
_t_encode_ms = (time.perf_counter() - _t_encode_start) * 1000
logger.info("Video response encoding (MP4 bytes): %.2f ms", _t_encode_ms)
return video_bytes
@staticmethod
def _apply_lora(lora_body: Any, gen_params: OmniDiffusionSamplingParams) -> None:
try: