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

enhancer

Provider-agnostic prompt orchestration for the streaming server.

Three operations the streaming server needs:

  • enhance — polish a user prompt (add cinematic detail, fix syntax)
  • auto_extend — generate a follow-on prompt for loop generation
  • rewrite — rewrite a seed prompt for a user-directed rewrite flow

All three share the same orchestration: pick a provider in priority order, submit an LLMRequest, fall back to the next provider on retryable errors, and surface a structured :class:LLMResponse back to the caller.

System prompts are loaded from system_prompt_dir on construction and can be hot-reloaded via :meth:PromptEnhancer.reload_system_prompts. The streaming server's management endpoint calls that method in response to a rewrite_seed_prompts_started frame.

fastvideo.entrypoints.streaming.prompt.enhancer.LLMMessage

LLMMessage(role: Literal['system', 'user', 'assistant'], content: str)

fastvideo.entrypoints.streaming.prompt.enhancer.LLMProvider

LLMProvider()

Provider interface every LLM adapter implements.

Providers are async-first because every built-in implementation talks to an HTTP API. Synchronous providers can wrap their call in asyncio.to_thread internally.

fastvideo.entrypoints.streaming.prompt.enhancer.LLMProvider.complete

complete(request: LLMRequest) -> LLMResponse

fastvideo.entrypoints.streaming.prompt.enhancer.LLMProviderError

LLMProviderError(message: str, *, retryable: bool = True)

Raised when an LLM provider fails a request.

retryable controls whether the enhancer falls back to the next provider. It is settable per-instance so the same exception type can describe retryable transport errors (5xx, 429) and non-retryable client errors (4xx auth/bad-request) without forcing a separate subclass for every status family.

fastvideo.entrypoints.streaming.prompt.enhancer.LLMRequest

LLMRequest(messages: list[LLMMessage], model: str, max_tokens: int | None = None, temperature: float | None = None, timeout_ms: int | None = None)

fastvideo.entrypoints.streaming.prompt.enhancer.LLMResponse

LLMResponse(content: str, provider: str, model: str, latency_ms: float, fallback_used: bool = False)

fastvideo.entrypoints.streaming.prompt.enhancer.init_logger

init_logger(name: str) -> _FastvideoLogger

The main purpose of this function is to ensure that loggers are retrieved in such a way that we can be sure the root fastvideo logger has already been configured.

fastvideo.entrypoints.streaming.prompt.enhancer.PromptOperation

PromptOperation()
Source
class PromptOperation(enum.Enum):
    ENHANCE = "enhance"
    AUTO_EXTEND = "auto_extend"
    REWRITE = "rewrite"

fastvideo.entrypoints.streaming.prompt.enhancer.PromptEnhancer

PromptEnhancer(*, providers: Sequence[LLMProvider], model: str, timeout_ms: int = 20000, temperature: float = 0.7, max_tokens: int | None = 256, system_prompt_dir: str | None = None)

Orchestrates prompt operations across a priority-ordered provider list with structured fallback + hot-reloadable system prompts.

Usage::

enhancer = PromptEnhancer(
    providers=[CerebrasProvider(), GroqProvider()],
    model="gpt-oss-120b",
    system_prompt_dir="/etc/fastvideo/prompts",
)
response = await enhancer.enhance("a fox running through snow")
Source
class PromptEnhancer:
    """Orchestrates prompt operations across a priority-ordered provider
    list with structured fallback + hot-reloadable system prompts.

Usage::

    enhancer = PromptEnhancer(
        providers=[CerebrasProvider(), GroqProvider()],
        model="gpt-oss-120b",
        system_prompt_dir="/etc/fastvideo/prompts",
    )
    response = await enhancer.enhance("a fox running through snow")
"""

def __init__(
    self,
    *,
    providers: Sequence[LLMProvider],
    model: str,
    timeout_ms: int = 20000,
    temperature: float = 0.7,
    max_tokens: int | None = 256,
    system_prompt_dir: str | None = None,
) -> None:
    if not providers:
        raise ValueError("PromptEnhancer requires at least one LLMProvider")
    self._providers = list(providers)
    self._model = model
    self._timeout_ms = timeout_ms
    self._temperature = temperature
    self._max_tokens = max_tokens
    self._system_prompt_dir = system_prompt_dir
    self._system_prompts = self._load_system_prompts()

@property
def providers(self) -> list[LLMProvider]:
    return list(self._providers)

def register_provider(self, provider: LLMProvider, *, priority: int = -1) -> None:
    """Insert an additional provider. ``priority=0`` makes it primary;
    ``priority=-1`` (default) appends as a fallback."""
    if priority < 0:
        self._providers.append(provider)
    else:
        self._providers.insert(priority, provider)

def reload_system_prompts(self) -> None:
    """Re-read the system prompt files from ``system_prompt_dir``.

    The streaming server exposes this via a management endpoint so
    operators can iterate on prompt templates without restarting
    workers.
    """
    self._system_prompts = self._load_system_prompts()
    logger.info("prompt enhancer: reloaded system prompts from %s", self._system_prompt_dir or "defaults")

async def enhance(self, prompt: str) -> LLMResponse:
    return await self._run(
        PromptOperation.ENHANCE,
        system=self._system_prompts.enhance,
        user=prompt,
    )

async def auto_extend(self, prior_prompts: Sequence[str]) -> LLMResponse:
    user = "\n".join(prior_prompts)
    return await self._run(
        PromptOperation.AUTO_EXTEND,
        system=self._system_prompts.auto_extend,
        user=user,
    )

async def rewrite(self, seed_prompt: str) -> LLMResponse:
    return await self._run(
        PromptOperation.REWRITE,
        system=self._system_prompts.rewrite,
        user=seed_prompt,
    )

async def _run(
    self,
    operation: PromptOperation,
    *,
    system: str,
    user: str,
) -> LLMResponse:
    request = LLMRequest(
        messages=[
            LLMMessage(role="system", content=system),
            LLMMessage(role="user", content=user),
        ],
        model=self._model,
        max_tokens=self._max_tokens,
        temperature=self._temperature,
        timeout_ms=self._timeout_ms,
    )
    last_error: LLMProviderError | None = None
    for idx, provider in enumerate(self._providers):
        try:
            response = await provider.complete(request)
            if idx > 0:
                # Mark the fallback flag without losing any other
                # response fields the provider populated.
                response = replace(response, fallback_used=True)
            return response
        except LLMProviderError as exc:
            logger.warning("prompt %s: provider %s failed: %s; trying next", operation.value, provider.name, exc)
            last_error = exc
            if not exc.retryable:
                break
    assert last_error is not None
    raise last_error

def _load_system_prompts(self) -> _SystemPrompts:
    if not self._system_prompt_dir:
        return _DEFAULT_SYSTEM_PROMPTS
    return _SystemPrompts(
        enhance=_read_prompt(self._system_prompt_dir, "enhance.txt", _DEFAULT_SYSTEM_PROMPTS.enhance),
        auto_extend=_read_prompt(self._system_prompt_dir, "auto_extend.txt", _DEFAULT_SYSTEM_PROMPTS.auto_extend),
        rewrite=_read_prompt(self._system_prompt_dir, "rewrite.txt", _DEFAULT_SYSTEM_PROMPTS.rewrite),
    )</code></pre></details>


fastvideo.entrypoints.streaming.prompt.enhancer.PromptEnhancer.register_provider


register_provider(provider: LLMProvider, *, priority: int = -1) -> None


Insert an additional provider. priority=0 makes it primary;
priority=-1 (default) appends as a fallback.


Source
def register_provider(self, provider: LLMProvider, *, priority: int = -1) -> None:
    """Insert an additional provider. ``priority=0`` makes it primary;
    ``priority=-1`` (default) appends as a fallback."""
    if priority < 0:
        self._providers.append(provider)
    else:
        self._providers.insert(priority, provider)


fastvideo.entrypoints.streaming.prompt.enhancer.PromptEnhancer.reload_system_prompts


reload_system_prompts() -> None


Re-read the system prompt files from system_prompt_dir.


The streaming server exposes this via a management endpoint so
operators can iterate on prompt templates without restarting
workers.


Source
def reload_system_prompts(self) -> None:
    """Re-read the system prompt files from ``system_prompt_dir``.

The streaming server exposes this via a management endpoint so
operators can iterate on prompt templates without restarting
workers.
&quot;&quot;&quot;
self._system_prompts = self._load_system_prompts()
logger.info(&quot;prompt enhancer: reloaded system prompts from %s&quot;, self._system_prompt_dir or &quot;defaults&quot;)</code></pre></details>


fastvideo.entrypoints.streaming.prompt.enhancer.PromptEnhancer.enhance


enhance(prompt: str) -> LLMResponse


Source
async def enhance(self, prompt: str) -> LLMResponse:
    return await self._run(
        PromptOperation.ENHANCE,
        system=self._system_prompts.enhance,
        user=prompt,
    )


fastvideo.entrypoints.streaming.prompt.enhancer.PromptEnhancer.auto_extend


auto_extend(prior_prompts: Sequence[str]) -> LLMResponse


Source
async def auto_extend(self, prior_prompts: Sequence[str]) -> LLMResponse:
    user = "\n".join(prior_prompts)
    return await self._run(
        PromptOperation.AUTO_EXTEND,
        system=self._system_prompts.auto_extend,
        user=user,
    )


fastvideo.entrypoints.streaming.prompt.enhancer.PromptEnhancer.rewrite


rewrite(seed_prompt: str) -> LLMResponse


Source
async def rewrite(self, seed_prompt: str) -> LLMResponse:
    return await self._run(
        PromptOperation.REWRITE,
        system=self._system_prompts.rewrite,
        user=seed_prompt,
    )

Was this page helpful?