Skip to content

prompt

Prompt pipeline for the streaming server.

  • :mod:providers — LLM backend abstraction + built-in adapters
  • :mod:enhancer — provider-agnostic enhance / auto-extend / rewrite operations on top of the provider layer

All of this is optional; the streaming server runs fine without it (PR 7.5's skeleton never invokes the enhancer). When the operator enables ServeConfig.streaming.prompt.enabled, the server routes each session_init_v2 curated prompt through enhance before the first segment.

Classes

fastvideo.entrypoints.streaming.prompt.LLMProvider

Bases: Protocol

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.LLMProviderError 

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

Bases: RuntimeError

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.

Source code in fastvideo/entrypoints/streaming/prompt/providers/base.py 
51
52
53
def __init__(self, message: str, *, retryable: bool = True) -> None:
    super().__init__(message)
    self.retryable = retryable

fastvideo.entrypoints.streaming.prompt.LLMTimeoutError 

LLMTimeoutError(message: str)

Bases: LLMProviderError

Raised when an LLM provider times out — always retryable.

Source code in fastvideo/entrypoints/streaming/prompt/providers/base.py 
59
60
def __init__(self, message: str) -> None:
    super().__init__(message, retryable=True)

fastvideo.entrypoints.streaming.prompt.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 code in fastvideo/entrypoints/streaming/prompt/enhancer.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
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()

Methods:

fastvideo.entrypoints.streaming.prompt.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 code in fastvideo/entrypoints/streaming/prompt/enhancer.py 
104
105
106
107
108
109
110
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.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 code in fastvideo/entrypoints/streaming/prompt/enhancer.py 
112
113
114
115
116
117
118
119
120
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")