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),
        )

`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.
    """
    self._system_prompts = self._load_system_prompts()
    logger.info("prompt enhancer: reloaded system prompts from %s", self._system_prompt_dir or "defaults")

`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,
    )