Module livekit.plugins.krisp.viva_filter

Krisp VIVA noise reduction audio filter for LiveKit Agents.

This module exposes :class:KrispVivaFilterFrameProcessor, a thin facade that forwards to one of two underlying FrameProcessor implementations:

  • The closed-source krisp_audio_livekit_internal wheel (default; authenticates via the LiveKit Cloud-managed JWT the agent framework hands to FrameProcessors through _on_credentials_updated).
  • A local license-mode wrapper around krisp_audio (selected when the user passes :class:KrispLicenseAuthProvider).

Note: isinstance(processor, SomeBackendInternal) will not match — the public class is the facade, not the backend it forwards to.

Classes

class KrispVivaFilterFrameProcessor (*,
auth_provider: LiveKitCloudAuthProvider | KrispLicenseAuthProvider | None = None,
model_path: str | None = None,
noise_suppression_level: int = 100,
frame_duration_ms: int | None = None,
sample_rate: int | None = None)
Expand source code 
class KrispVivaFilterFrameProcessor(rtc.FrameProcessor[rtc.AudioFrame]):
    """FrameProcessor for Krisp noise reduction.

    Thin facade over two backend FrameProcessor implementations: the
    LiveKit Cloud-bundled wheel (default) and a local wrapper around the
    public ``krisp_audio`` wheel (selected via
    :class:`KrispLicenseAuthProvider`).

    Example:
        ```python
        from livekit.agents import room_io
        from livekit.plugins import krisp

        # Default: uses LiveKit Cloud auth + bundled model.
        processor = krisp.KrispVivaFilterFrameProcessor()

        # Or, explicit Krisp-direct auth with a license + model file.
        processor = krisp.KrispVivaFilterFrameProcessor(
            auth_provider=krisp.auth.krisp_license(
                license_key="...",
                model_path="/path/to/model.kef",
            ),
        )

        await session.start(
            agent=MyAgent(),
            room=ctx.room,
            room_options=room_io.RoomOptions(
                audio_input=room_io.AudioInputOptions(
                    noise_cancellation=processor,
                ),
            ),
        )
        ```
    """

    def __init__(
        self,
        *,
        auth_provider: LiveKitCloudAuthProvider | KrispLicenseAuthProvider | None = None,
        model_path: str | None = None,
        noise_suppression_level: int = 100,
        frame_duration_ms: int | None = None,
        sample_rate: int | None = None,
    ) -> None:
        """Initialize the Krisp frame processor.

        Args:
            noise_suppression_level: Noise suppression level (0-100, default: 100).
            auth_provider: Authentication provider. Defaults to
                :class:`LiveKitCloudAuthProvider` (LiveKit Cloud auth + bundled
                model). Pass :class:`KrispLicenseAuthProvider` to use a Krisp
                license key + ``.kef`` model file directly.
            model_path: **Deprecated.** Use
                ``auth_provider=KrispLicenseAuthProvider(model_path=...)``
                instead. Path to the Krisp model file (``.kef``).
            frame_duration_ms: **Deprecated.** The processor now buffers input
                frames of any size automatically, so this no longer needs to be
                set. Frame duration in milliseconds (10, 15, 20, 30, or 32).
            sample_rate: **Deprecated.** The processor now adapts to the input
                sample rate automatically, so this no longer needs to be set.
                Sample rate in Hz.

        Raises:
            RuntimeError: If the chosen backend wheel is not installed.
            ValueError: If ``frame_duration_ms`` is not supported, or — for
                :class:`KrispLicenseAuthProvider` — if ``model_path`` is
                missing or does not have a ``.kef`` extension.
            FileNotFoundError: If the license-mode model file does not exist.
        """
        if frame_duration_ms is not None or sample_rate is not None:
            global _FRAME_PARAMS_DEPRECATION_SHOWN
            if not _FRAME_PARAMS_DEPRECATION_SHOWN:
                warnings.warn(
                    "Passing `sample_rate` / `frame_duration_ms` to "
                    "KrispVivaFilterFrameProcessor is deprecated. The processor "
                    "now adapts to the input sample rate and frame size "
                    "automatically.",
                    DeprecationWarning,
                    stacklevel=2,
                )
                _FRAME_PARAMS_DEPRECATION_SHOWN = True

        provider = _resolve_auth_provider(auth_provider, model_path)
        self._inner = _build_inner(
            provider,
            noise_suppression_level=noise_suppression_level,
            frame_duration_ms=frame_duration_ms if frame_duration_ms is not None else 10,
            sample_rate=sample_rate,
        )

    # ----- FrameProcessor hooks: forward to the inner processor -------------

    def _process(self, frame: rtc.AudioFrame) -> rtc.AudioFrame:
        return self._inner._process(frame)

    def _on_credentials_updated(self, *, token: str, url: str) -> None:
        self._inner._on_credentials_updated(token=token, url=url)

    def _on_stream_info_updated(
        self,
        *,
        room_name: str,
        participant_identity: str,
        publication_sid: str,
    ) -> None:
        self._inner._on_stream_info_updated(
            room_name=room_name,
            participant_identity=participant_identity,
            publication_sid=publication_sid,
        )

    def _close(self) -> None:
        self._inner._close()

    @property
    def enabled(self) -> bool:
        return self._inner.enabled

    @enabled.setter
    def enabled(self, value: bool) -> None:
        self._inner.enabled = value

    @property
    def noise_suppression_level(self) -> float:
        """Current noise suppression level (0-100)."""
        return self._inner.noise_suppression_level

    @noise_suppression_level.setter
    def noise_suppression_level(self, value: float) -> None:
        """Adjust the noise suppression level (0-100) at runtime."""
        self._inner.noise_suppression_level = value

    # ----- Backwards-compat shims ------------------------------------------

    def process(self, frame: rtc.AudioFrame) -> rtc.AudioFrame:
        """Public method that calls _process (for backward compatibility)."""
        return self._process(frame)

    def enable(self) -> None:
        """Enable noise filtering."""
        self.enabled = True

    def disable(self) -> None:
        """Disable noise filtering (audio will pass through unmodified)."""
        self.enabled = False

    @property
    def is_enabled(self) -> bool:
        """Check if filtering is currently enabled (backward compatibility)."""
        return self.enabled

    def close(self) -> None:
        """Clean up processor session resources (public method for backward compatibility)."""
        self._close()

    def __enter__(self) -> KrispVivaFilterFrameProcessor:
        """Context manager entry."""
        return self

    def __exit__(
        self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: Any
    ) -> Literal[False]:
        """Context manager exit - clean up session."""
        self.close()
        return False

FrameProcessor for Krisp noise reduction.

Thin facade over two backend FrameProcessor implementations: the LiveKit Cloud-bundled wheel (default) and a local wrapper around the public krisp_audio wheel (selected via :class:KrispLicenseAuthProvider).

Example

from livekit.agents import room_io
from livekit.plugins import krisp

# Default: uses LiveKit Cloud auth + bundled model.
processor = krisp.KrispVivaFilterFrameProcessor()

# Or, explicit Krisp-direct auth with a license + model file.
processor = krisp.KrispVivaFilterFrameProcessor(
    auth_provider=krisp.auth.krisp_license(
        license_key="...",
        model_path="/path/to/model.kef",
    ),
)

await session.start(
    agent=MyAgent(),
    room=ctx.room,
    room_options=room_io.RoomOptions(
        audio_input=room_io.AudioInputOptions(
            noise_cancellation=processor,
        ),
    ),
)

Initialize the Krisp frame processor.

Args

noise_suppression_level
Noise suppression level (0-100, default: 100).
auth_provider
Authentication provider. Defaults to :class:LiveKitCloudAuthProvider (LiveKit Cloud auth + bundled model). Pass :class:KrispLicenseAuthProvider to use a Krisp license key + .kef model file directly.
model_path
Deprecated. Use auth_provider=KrispLicenseAuthProvider(model_path=...) instead. Path to the Krisp model file (.kef).
frame_duration_ms
Deprecated. The processor now buffers input frames of any size automatically, so this no longer needs to be set. Frame duration in milliseconds (10, 15, 20, 30, or 32).
sample_rate
Deprecated. The processor now adapts to the input sample rate automatically, so this no longer needs to be set. Sample rate in Hz.

Raises

RuntimeError
If the chosen backend wheel is not installed.
ValueError
If frame_duration_ms is not supported, or — for :class:KrispLicenseAuthProvider — if model_path is missing or does not have a .kef extension.
FileNotFoundError
If the license-mode model file does not exist.

Ancestors

Instance variables

prop enabled : bool
Expand source code 
@property
def enabled(self) -> bool:
    return self._inner.enabled
prop is_enabled : bool
Expand source code 
@property
def is_enabled(self) -> bool:
    """Check if filtering is currently enabled (backward compatibility)."""
    return self.enabled

Check if filtering is currently enabled (backward compatibility).

prop noise_suppression_level : float
Expand source code 
@property
def noise_suppression_level(self) -> float:
    """Current noise suppression level (0-100)."""
    return self._inner.noise_suppression_level

Current noise suppression level (0-100).

Methods

def close(self) ‑> None
Expand source code 
def close(self) -> None:
    """Clean up processor session resources (public method for backward compatibility)."""
    self._close()

Clean up processor session resources (public method for backward compatibility).

def disable(self) ‑> None
Expand source code 
def disable(self) -> None:
    """Disable noise filtering (audio will pass through unmodified)."""
    self.enabled = False

Disable noise filtering (audio will pass through unmodified).

def enable(self) ‑> None
Expand source code 
def enable(self) -> None:
    """Enable noise filtering."""
    self.enabled = True

Enable noise filtering.

def process(self, frame: rtc.AudioFrame) ‑> AudioFrame
Expand source code 
def process(self, frame: rtc.AudioFrame) -> rtc.AudioFrame:
    """Public method that calls _process (for backward compatibility)."""
    return self._process(frame)

Public method that calls _process (for backward compatibility).