Skip to content

safety

Optional prompt safety filter.

Uses a fastText classifier to score prompts against a banned-content rubric. Only loaded when ServeConfig.streaming.safety.enabled is True and fastText is installed — users who don't need it see no runtime cost.

Install: pip install fastvideo[prompt-safety] (ships fasttext as an optional extra) or install fasttext directly.

Classes

fastvideo.entrypoints.streaming.prompt.safety.PromptSafetyFilter

PromptSafetyFilter(*, classifier_path: str | None, enabled: bool = True, block_threshold: float = 0.5)

Minimal fastText-backed prompt safety filter.

Loads the classifier lazily on first use so the streaming server can construct the filter eagerly at startup without paying the model-load cost when safety is disabled.

Source code in fastvideo/entrypoints/streaming/prompt/safety.py
def __init__(
    self,
    *,
    classifier_path: str | None,
    enabled: bool = True,
    block_threshold: float = 0.5,
) -> None:
    self._classifier_path = classifier_path
    self._enabled = enabled
    self._block_threshold = block_threshold
    self._model: Any | None = None
    self._load_attempted = False
    self._load_lock = threading.Lock()

fastvideo.entrypoints.streaming.prompt.safety.SafetyDecision

Bases: Enum

Attributes

fastvideo.entrypoints.streaming.prompt.safety.SafetyDecision.UNAVAILABLE class-attribute instance-attribute
UNAVAILABLE = 'unavailable'

Returned when the classifier can't run (not configured, fastText missing). Safety is opt-in; the server treats UNAVAILABLE as ALLOW but logs it so operators know the filter is off.

Functions

fastvideo.entrypoints.streaming.prompt.safety.first_blocked

first_blocked(filter_: PromptSafetyFilter, prompts: list[str]) -> SafetyResult | None

Return the first prompt the filter blocks, or None.

Source code in fastvideo/entrypoints/streaming/prompt/safety.py
def first_blocked(
    filter_: PromptSafetyFilter,
    prompts: list[str],
) -> SafetyResult | None:
    """Return the first prompt the filter blocks, or ``None``."""
    for prompt in prompts:
        result = filter_.classify(prompt)
        if result.decision is SafetyDecision.BLOCK:
            return result
    return None