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

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.

fastvideo.entrypoints.streaming.prompt.safety.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.safety.SafetyDecision

SafetyDecision()
Source
class SafetyDecision(enum.Enum):
    ALLOW = "allow"
    BLOCK = "block"
    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."""

fastvideo.entrypoints.streaming.prompt.safety.SafetyDecision.UNAVAILABLE

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.

Source
UNAVAILABLE = "unavailable"

fastvideo.entrypoints.streaming.prompt.safety.SafetyResult

SafetyResult(prompt: str, decision: SafetyDecision, score: float = 0.0, label: str | None = None, reason: str | None = None)
Source
@dataclass
class SafetyResult:
    prompt: str
    decision: SafetyDecision
    score: float = 0.0
    label: str | None = None
    reason: str | None = None

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
class PromptSafetyFilter:
    """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.
"""

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

@property
def enabled(self) -> bool:
    return self._enabled and self._classifier_path is not None

def classify(self, prompt: str) -> SafetyResult:
    if not self.enabled:
        return SafetyResult(
            prompt=prompt,
            decision=SafetyDecision.UNAVAILABLE,
            reason="safety filter not enabled",
        )
    model = self._ensure_loaded()
    if model is None:
        return SafetyResult(
            prompt=prompt,
            decision=SafetyDecision.UNAVAILABLE,
            reason="fastText model unavailable",
        )
    try:
        labels, probs = model.predict(prompt.replace("\n", " "), k=1)
    except Exception as exc:  # pragma: no cover - defensive
        logger.warning("safety: classifier failed: %s", exc)
        return SafetyResult(
            prompt=prompt,
            decision=SafetyDecision.UNAVAILABLE,
            reason=f"classifier error: {exc}",
        )
    label = labels[0].removeprefix("__label__") if labels else None
    score = float(probs[0]) if len(probs) else 0.0
    decision = (SafetyDecision.BLOCK if
                (label == "unsafe" and score >= self._block_threshold) else SafetyDecision.ALLOW)
    return SafetyResult(
        prompt=prompt,
        decision=decision,
        score=score,
        label=label,
    )

def _ensure_loaded(self) -> Any | None:
    if self._model is not None:
        return self._model
    if self._load_attempted:
        return None
    with self._load_lock:
        if self._model is not None:
            return self._model
        if self._load_attempted:
            return None
        self._load_attempted = True
        if self._classifier_path is None:
            return None
        try:
            import fasttext  # type: ignore[import-not-found]
        except ImportError:
            logger.warning("safety: fasttext not installed; safety filter disabled. "
                           "Install fastvideo[prompt-safety] to enable.")
            return None
        try:
            self._model = fasttext.load_model(self._classifier_path)
        except Exception as exc:  # pragma: no cover - requires real model
            logger.warning("safety: failed to load %s: %s", self._classifier_path, exc)
            return None
        return self._model</code></pre></details>


fastvideo.entrypoints.streaming.prompt.safety.PromptSafetyFilter.classify


classify(prompt: str) -> SafetyResult


Source
def classify(self, prompt: str) -> SafetyResult:
    if not self.enabled:
        return SafetyResult(
            prompt=prompt,
            decision=SafetyDecision.UNAVAILABLE,
            reason="safety filter not enabled",
        )
    model = self._ensure_loaded()
    if model is None:
        return SafetyResult(
            prompt=prompt,
            decision=SafetyDecision.UNAVAILABLE,
            reason="fastText model unavailable",
        )
    try:
        labels, probs = model.predict(prompt.replace("\n", " "), k=1)
    except Exception as exc:  # pragma: no cover - defensive
        logger.warning("safety: classifier failed: %s", exc)
        return SafetyResult(
            prompt=prompt,
            decision=SafetyDecision.UNAVAILABLE,
            reason=f"classifier error: {exc}",
        )
    label = labels[0].removeprefix("__label__") if labels else None
    score = float(probs[0]) if len(probs) else 0.0
    decision = (SafetyDecision.BLOCK if
                (label == "unsafe" and score >= self._block_threshold) else SafetyDecision.ALLOW)
    return SafetyResult(
        prompt=prompt,
        decision=decision,
        score=score,
        label=label,
    )


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

Was this page helpful?