Skip to content
FastVideo
Esc
↑↓navigate↵open⌘Jpreview
On this page

serving_engine

Shared asynchronous execution substrate for OpenAI-compatible routes.

fastvideo.entrypoints.openai.serving_engine.GenerationRequest

GenerationRequest(prompt: str | list[str] | None = None, negative_prompt: str | None = None, inputs: InputConfig = InputConfig(), sampling: SamplingConfig = SamplingConfig(), runtime: RequestRuntimeConfig = RequestRuntimeConfig(), output: OutputConfig = OutputConfig(), stage_overrides: dict[str, Any] = dict(), state: ContinuationState | None = None, plan: GenerationPlan | None = None, extensions: dict[str, Any] = dict())

fastvideo.entrypoints.openai.serving_engine.VideoGenerationRequest

VideoGenerationRequest()

OpenAI/vLLM-Omni-compatible video generation request.

Model-specific parameters belong in extra_params. FastVideo keeps the legacy input_reference and reference_url fields for clients that predate vLLM-Omni's typed reference objects.

fastvideo.entrypoints.openai.serving_engine.VideoGenerationRequest.validate_prompt

validate_prompt(value: str) -> str

fastvideo.entrypoints.openai.serving_engine.VideoGenerationRequest.resolve_video_params

resolve_video_params() -> VideoParams

Resolve top-level, nested, and size dimensions like vLLM-Omni.

fastvideo.entrypoints.openai.serving_engine.ServingGenerator

ServingGenerator()
Source
class ServingGenerator(Protocol):

def generate(self, request: GenerationRequest) -> Any:
    ...

def shutdown(self) -> None:
    ...</code></pre></details>


fastvideo.entrypoints.openai.serving_engine.ServingGenerator.generate


generate(request: GenerationRequest) -> Any


Source
def generate(self, request: GenerationRequest) -> Any:
    ...


fastvideo.entrypoints.openai.serving_engine.ServingGenerator.shutdown


shutdown() -> None


Source
def shutdown(self) -> None:
    ...


fastvideo.entrypoints.openai.serving_engine.OpenAIServingEngine


OpenAIServingEngine(generator: ServingGenerator, video_request_validator: Callable[[VideoGenerationRequest], None] | None = None)


Own generator lifecycle and serialize access to its mutable pipeline.


FastVideo pipelines contain request-mutated sampling state and some LoRA
implementations merge weights in place. Running two Python threads through
one pipeline is therefore unsafe even if the HTTP layer accepts requests
concurrently. This engine gives every OpenAI route one model-agnostic async
entrypoint while preserving that invariant. A future scheduler can replace
the lock without changing the transport contract.


Source
class OpenAIServingEngine:
    """Own generator lifecycle and serialize access to its mutable pipeline.

FastVideo pipelines contain request-mutated sampling state and some LoRA
implementations merge weights in place. Running two Python threads through
one pipeline is therefore unsafe even if the HTTP layer accepts requests
concurrently. This engine gives every OpenAI route one model-agnostic async
entrypoint while preserving that invariant. A future scheduler can replace
the lock without changing the transport contract.
&quot;&quot;&quot;

def __init__(self,
             generator: ServingGenerator,
             video_request_validator: Callable&#91;&#91;VideoGenerationRequest&#93;, None&#93; | None = None) -&gt; None:
    self._generator = generator
    self._video_request_validator = video_request_validator
    self._generation_lock = asyncio.Lock()
    self._closed = False
    self._unhealthy_reason: str | None = None

@property
def generator(self) -&gt; ServingGenerator:
    return self._generator

def validate_video_request(self, request: VideoGenerationRequest) -&gt; None:
    if self._video_request_validator is not None:
        self._video_request_validator(request)

@property
def closed(self) -&gt; bool:
    return self._closed

@property
def healthy(self) -&gt; bool:
    if self._closed or self._unhealthy_reason is not None:
        return False
    executor = getattr(self._generator, &quot;executor&quot;, None)
    workers = getattr(executor, &quot;workers&quot;, None)
    if workers is None:
        return True
    return bool(workers) and all(worker.proc.is_alive() for worker in workers)

@property
def unhealthy_reason(self) -&gt; str | None:
    if self._unhealthy_reason is not None:
        return self._unhealthy_reason
    if not self.healthy:
        return &quot;one or more generation workers are not alive&quot;
    return None

async def generate(
    self,
    request: GenerationRequest,
    *,
    on_start: Callable&#91;&#91;&#93;, Awaitable&#91;None&#93;&#93; | None = None,
) -&gt; Any:
    &quot;&quot;&quot;Generate one typed request without blocking the event loop.&quot;&quot;&quot;
    return await self.run_serialized(self._generator.generate, request, on_start=on_start)

async def run_serialized(
    self,
    function: Callable&#91;..., _T&#93;,
    *args: Any,
    on_start: Callable&#91;&#91;&#93;, Awaitable&#91;None&#93;&#93; | None = None,
    **kwargs: Any,
) -&gt; _T:
    &quot;&quot;&quot;Run a synchronous pipeline operation under the serving lock.&quot;&quot;&quot;
    if self._closed:
        raise RuntimeError(&quot;FastVideo serving engine is shutting down&quot;)
    async with self._generation_lock:
        if self._closed:
            raise RuntimeError(&quot;FastVideo serving engine is shutting down&quot;)
        if on_start is not None:
            await on_start()
        worker = asyncio.create_task(asyncio.to_thread(function, *args, **kwargs))
        try:
            return await asyncio.shield(worker)
        except asyncio.CancelledError:
            # Python cannot stop a running worker thread. Keep the lock
            # until the pipeline call really exits so cancellation cannot
            # expose mutable model state to a second request.
            await self._wait_after_cancellation(worker)
            raise
        except (BrokenPipeError, EOFError) as error:
            self._unhealthy_reason = str(error)
            raise

async def run_async_serialized(self, function: Callable&#91;&#91;&#93;, Awaitable&#91;_T&#93;&#93;) -&gt; _T:
    &quot;&quot;&quot;Run an async operation under the same pipeline lock.&quot;&quot;&quot;
    if self._closed:
        raise RuntimeError(&quot;FastVideo serving engine is shutting down&quot;)
    async with self._generation_lock:
        if self._closed:
            raise RuntimeError(&quot;FastVideo serving engine is shutting down&quot;)
        worker: asyncio.Future&#91;_T&#93; = asyncio.ensure_future(function())
        try:
            return await asyncio.shield(worker)
        except asyncio.CancelledError:
            await self._wait_after_cancellation(worker)
            raise

@staticmethod
async def _wait_after_cancellation(worker: asyncio.Future&#91;Any&#93;) -&gt; None:
    &quot;&quot;&quot;Keep waiting for an uninterruptible worker despite repeated cancellation.&quot;&quot;&quot;
    while not worker.done():
        try:
            await asyncio.shield(worker)
        except asyncio.CancelledError:
            continue
        except Exception:
            break
    if not worker.cancelled():
        worker.exception()

async def shutdown(self) -&gt; None:
    &quot;&quot;&quot;Stop accepting requests and release the generator after in-flight work.&quot;&quot;&quot;
    self._closed = True
    async with self._generation_lock:
        await asyncio.to_thread(self._generator.shutdown)</code></pre></details>


fastvideo.entrypoints.openai.serving_engine.OpenAIServingEngine.validate_video_request


validate_video_request(request: VideoGenerationRequest) -> None


Source
def validate_video_request(self, request: VideoGenerationRequest) -> None:
    if self._video_request_validator is not None:
        self._video_request_validator(request)


fastvideo.entrypoints.openai.serving_engine.OpenAIServingEngine.generate


generate(request: GenerationRequest, *, on_start: Callable[[], Awaitable[None]] | None = None) -> Any


Generate one typed request without blocking the event loop.


Source
async def generate(
    self,
    request: GenerationRequest,
    *,
    on_start: Callable[[], Awaitable[None]] | None = None,
) -> Any:
    """Generate one typed request without blocking the event loop."""
    return await self.run_serialized(self._generator.generate, request, on_start=on_start)


fastvideo.entrypoints.openai.serving_engine.OpenAIServingEngine.run_serialized


run_serialized(function: Callable[..., _T], *args: Any, on_start: Callable[[], Awaitable[None]] | None = None, **kwargs: Any) -> _T


Run a synchronous pipeline operation under the serving lock.


Source
async def run_serialized(
    self,
    function: Callable[..., _T],
    *args: Any,
    on_start: Callable[[], Awaitable[None]] | None = None,
    **kwargs: Any,
) -> _T:
    """Run a synchronous pipeline operation under the serving lock."""
    if self._closed:
        raise RuntimeError("FastVideo serving engine is shutting down")
    async with self._generation_lock:
        if self._closed:
            raise RuntimeError("FastVideo serving engine is shutting down")
        if on_start is not None:
            await on_start()
        worker = asyncio.create_task(asyncio.to_thread(function, *args, **kwargs))
        try:
            return await asyncio.shield(worker)
        except asyncio.CancelledError:
            # Python cannot stop a running worker thread. Keep the lock
            # until the pipeline call really exits so cancellation cannot
            # expose mutable model state to a second request.
            await self._wait_after_cancellation(worker)
            raise
        except (BrokenPipeError, EOFError) as error:
            self._unhealthy_reason = str(error)
            raise


fastvideo.entrypoints.openai.serving_engine.OpenAIServingEngine.run_async_serialized


run_async_serialized(function: Callable[[], Awaitable[_T]]) -> _T


Run an async operation under the same pipeline lock.


Source
async def run_async_serialized(self, function: Callable[[], Awaitable[_T]]) -> _T:
    """Run an async operation under the same pipeline lock."""
    if self._closed:
        raise RuntimeError("FastVideo serving engine is shutting down")
    async with self._generation_lock:
        if self._closed:
            raise RuntimeError("FastVideo serving engine is shutting down")
        worker: asyncio.Future[_T] = asyncio.ensure_future(function())
        try:
            return await asyncio.shield(worker)
        except asyncio.CancelledError:
            await self._wait_after_cancellation(worker)
            raise


fastvideo.entrypoints.openai.serving_engine.OpenAIServingEngine.shutdown


shutdown() -> None


Stop accepting requests and release the generator after in-flight work.


Source
async def shutdown(self) -> None:
    """Stop accepting requests and release the generator after in-flight work."""
    self._closed = True
    async with self._generation_lock:
        await asyncio.to_thread(self._generator.shutdown)

Was this page helpful?