Merge pull request #4022 from krispai/krisp-viva-vad-support

Draft Implementation for Krisp VIVA VAD.

changelog/4022.changed.md

@@ -0,0 +1 @@
+- Added `KrispVivaVadAnalyzer` for Voice Activity Detection using the Krisp VIVA SDK (requires `krisp_audio`).

examples/foundational/07p-interruptible-krisp-viva.py

@@ -30,6 +30,7 @@ from loguru import logger
 from pipecat.audio.filters.krisp_viva_filter import KrispVivaFilter
 from pipecat.audio.turn.krisp_viva_turn import KrispVivaTurn
 from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.audio.vad.vad_analyzer import VADParams
 from pipecat.frames.frames import LLMRunFrame
 from pipecat.metrics.metrics import TurnMetricsData
 from pipecat.observers.loggers.metrics_log_observer import MetricsLogObserver
@@ -63,17 +64,20 @@ transport_params = {
     "daily": lambda: DailyParams(
         audio_in_enabled=True,
         audio_out_enabled=True,
-        audio_in_filter=krisp_viva_filter,
+        vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),  # or KrispVivaVadAnalyzer
+        audio_in_filter=KrispVivaFilter(),
     ),
     "twilio": lambda: FastAPIWebsocketParams(
         audio_in_enabled=True,
         audio_out_enabled=True,
-        audio_in_filter=krisp_viva_filter,
+        vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),  # or KrispVivaVadAnalyzer
+        audio_in_filter=KrispVivaFilter(),
     ),
     "webrtc": lambda: TransportParams(
         audio_in_enabled=True,
         audio_out_enabled=True,
-        audio_in_filter=krisp_viva_filter,
+        vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),  # or KrispVivaVadAnalyzer
+        audio_in_filter=KrispVivaFilter(),
     ),
 }
 

src/pipecat/audio/vad/krisp_viva_vad.py

@@ -0,0 +1,217 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Krisp Voice Activity Detection (VAD) implementation for Pipecat.
+
+This module provides a VAD analyzer based on the Krisp VIVA SDK,
+which can detect voice activity in audio streams with high accuracy.
+Supports 8kHz, 16kHz, 32kHz, 44.1kHz and 48kHz sample rates.
+"""
+
+import os
+from typing import Optional
+
+import numpy as np
+from loguru import logger
+
+from pipecat.audio.krisp_instance import (
+    KrispVivaSDKManager,
+    int_to_krisp_frame_duration,
+    int_to_krisp_sample_rate,
+)
+from pipecat.audio.vad.vad_analyzer import VADAnalyzer, VADParams
+
+try:
+    import krisp_audio
+except ModuleNotFoundError as e:
+    logger.error(f"Exception: {e}")
+    logger.error("In order to use KrispVivaVADAnalyzer, you need to install krisp_audio.")
+    raise Exception(f"Missing module: {e}")
+
+
+class KrispVivaVadAnalyzer(VADAnalyzer):
+    """Voice Activity Detection analyzer using the Krisp VIVA SDK."""
+
+    def __init__(
+        self,
+        *,
+        model_path: Optional[str] = None,
+        frame_duration: int = 10,
+        sample_rate: Optional[int] = None,
+        params: Optional[VADParams] = None,
+    ):
+        """Initialize the Krisp VIVA VAD analyzer.
+
+        Args:
+            model_path: Path to the Krisp model file (.kef extension).
+                If None, uses KRISP_VIVA_VAD_MODEL_PATH environment variable.
+            frame_duration: Frame duration in milliseconds (default: 10ms).
+            sample_rate: Audio sample rate (must be 8000, 16000, 32000, 44100 or 48000 Hz).
+                If None, will be set later.
+            params: VAD parameters for detection configuration.
+
+        Raises:
+            ValueError: If model_path is not provided and KRISP_VIVA_VAD_MODEL_PATH is not set.
+            Exception: If model file doesn't have .kef extension.
+            FileNotFoundError: If model file doesn't exist.
+        """
+        super().__init__(sample_rate=sample_rate, params=params)
+
+        logger.debug("Loading Krisp VIVA VAD model...")
+
+        try:
+            # Set model path, checking environment if not specified
+            if model_path:
+                self._model_path = model_path
+            else:
+                self._model_path = os.getenv("KRISP_VIVA_VAD_MODEL_PATH")
+                if not self._model_path:
+                    logger.error(
+                        "Model path is not provided and KRISP_VIVA_VAD_MODEL_PATH is not set."
+                    )
+                    raise ValueError("Model path for KrispVivaVADAnalyzer must be provided.")
+
+            if not self._model_path.endswith(".kef"):
+                raise Exception("Model is expected with .kef extension")
+
+            if not os.path.isfile(self._model_path):
+                raise FileNotFoundError(f"Model file not found: {self._model_path}")
+
+            self._session = None
+            self._frame_duration_ms = frame_duration
+            self._samples_per_frame = None
+            # Calculate samples per frame if sample_rate is provided
+            if sample_rate is not None:
+                self._samples_per_frame = int((sample_rate * frame_duration) / 1000)
+
+            # Acquire SDK reference (will initialize on first call)
+            KrispVivaSDKManager.acquire()
+
+            logger.debug("Loaded Krisp VIVA VAD")
+
+        except Exception:
+            # If initialization fails, release the SDK reference
+            KrispVivaSDKManager.release()
+            raise
+
+    def _create_session(self, sample_rate: int, frame_duration: int):
+        """Create a Krisp VAD session with a specific sample rate.
+
+        Args:
+            sample_rate: Sample rate for the session
+            frame_duration: Frame duration in milliseconds
+
+        Returns:
+            Krisp VAD session instance
+
+        Raises:
+            RuntimeError: If session creation fails
+        """
+        try:
+            model_info = krisp_audio.ModelInfo()
+            model_info.path = self._model_path
+
+            vad_cfg = krisp_audio.VadSessionConfig()
+            vad_cfg.inputSampleRate = int_to_krisp_sample_rate(sample_rate)
+            vad_cfg.inputFrameDuration = int_to_krisp_frame_duration(frame_duration)
+            vad_cfg.modelInfo = model_info
+
+            self._samples_per_frame = int((sample_rate * frame_duration) / 1000)
+            session = krisp_audio.VadFloat.create(vad_cfg)
+            return session
+        except Exception as e:
+            logger.error(f"Failed to create Krisp VAD session: {e}", exc_info=True)
+            raise RuntimeError(f"Failed to create Krisp VAD session: {e}") from e
+
+    def set_sample_rate(self, sample_rate: int):
+        """Set the sample rate for audio processing.
+
+        Args:
+            sample_rate: Audio sample rate (must be 8000, 16000, 32000 or 48000 Hz).
+
+        Raises:
+            ValueError: If sample rate is not 8000, 16000, 32000 or 48000 Hz.
+            RuntimeError: If VAD session creation fails.
+        """
+        if (
+            sample_rate != 48000
+            and sample_rate != 44100
+            and sample_rate != 32000
+            and sample_rate != 16000
+            and sample_rate != 8000
+        ):
+            raise ValueError(
+                f"Krisp VIVA VAD sample rate needs to be 8000, 16000, 32000, 44100 or 48000 (sample rate: {sample_rate})"
+            )
+
+        # Create or recreate session with new sample rate
+        try:
+            self._session = self._create_session(sample_rate, self._frame_duration_ms)
+        except Exception as e:
+            logger.error(f"Failed to set sample rate: {e}", exc_info=True)
+            raise RuntimeError(f"Failed to create Krisp VAD session: {e}") from e
+
+        super().set_sample_rate(sample_rate)
+
+    def num_frames_required(self) -> int:
+        """Get the number of audio frames required for analysis.
+
+        Returns:
+            Number of frames (samples) needed for VAD processing based on
+            current sample rate and frame duration.
+        """
+        # If already calculated from session creation, return it
+        if self._samples_per_frame is not None:
+            return self._samples_per_frame
+
+        # Calculate from current sample rate if available
+        if self.sample_rate > 0:
+            return int((self.sample_rate * self._frame_duration_ms) / 1000)
+
+        # Fallback: calculate from initial sample rate if provided
+        if self._init_sample_rate is not None:
+            return int((self._init_sample_rate * self._frame_duration_ms) / 1000)
+
+        # Default fallback: assume 16kHz @ 10ms = 160 samples
+        return int((16000 * self._frame_duration_ms) / 1000)
+
+    def voice_confidence(self, buffer) -> float:
+        """Calculate voice activity confidence for the given audio buffer.
+
+        Args:
+            buffer: Audio buffer to analyze (bytes, int16 format).
+
+        Returns:
+            Voice confidence score between 0.0 and 1.0.
+        """
+        if self._session is None:
+            logger.warning("VAD session not initialized. Cannot process audio.")
+            return 0.0
+
+        try:
+            # Convert bytes buffer to float32 numpy array
+            # Buffer is int16 (2 bytes per sample), need to convert to float32
+            audio_int16 = np.frombuffer(buffer, dtype=np.int16)
+            # Normalize to [-1.0, 1.0] range
+            audio_float32 = audio_int16.astype(np.float32) / 32768.0
+
+            # Process through VAD session
+            voice_probability = self._session.process(audio_float32)
+
+            return voice_probability
+
+        except Exception as e:
+            logger.error(f"Error analyzing audio with Krisp VIVA VAD: {e}", exc_info=True)
+            return 0.0
+
+    async def cleanup(self):
+        """Cleanup analyzer resources."""
+        try:
+            self._session = None
+            KrispVivaSDKManager.release()
+        except Exception:
+            # Ignore errors during cleanup
+            pass

tests/test_krisp_viva_vad.py

@@ -0,0 +1,442 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Unit tests for KrispVivaVadAnalyzer."""
+
+import asyncio
+import os
+import sys
+import tempfile
+import unittest
+from unittest.mock import MagicMock, patch
+
+import numpy as np
+
+# Mock package version check before importing pipecat
+# This allows tests to run in development mode without installed package
+_version_patcher = patch("importlib.metadata.version", return_value="0.0.0-dev")
+_version_patcher.start()
+
+# Mock krisp_audio module BEFORE any pipecat imports
+# This allows tests to run without krisp_audio installed
+mock_krisp_audio = MagicMock()
+mock_krisp_audio.SamplingRate.Sr8000Hz = 8000
+mock_krisp_audio.SamplingRate.Sr16000Hz = 16000
+mock_krisp_audio.SamplingRate.Sr24000Hz = 24000
+mock_krisp_audio.SamplingRate.Sr32000Hz = 32000
+mock_krisp_audio.SamplingRate.Sr44100Hz = 44100
+mock_krisp_audio.SamplingRate.Sr48000Hz = 48000
+mock_krisp_audio.FrameDuration.Fd10ms = "10ms"
+mock_krisp_audio.FrameDuration.Fd15ms = "15ms"
+mock_krisp_audio.FrameDuration.Fd20ms = "20ms"
+mock_krisp_audio.FrameDuration.Fd30ms = "30ms"
+mock_krisp_audio.FrameDuration.Fd32ms = "32ms"
+
+# Install the mock in sys.modules before importing
+sys.modules["krisp_audio"] = mock_krisp_audio
+
+# Mock pipecat_ai_krisp package
+mock_pipecat_krisp = MagicMock()
+sys.modules["pipecat_ai_krisp"] = mock_pipecat_krisp
+sys.modules["pipecat_ai_krisp.audio"] = MagicMock()
+sys.modules["pipecat_ai_krisp.audio.krisp_processor"] = MagicMock()
+
+# Now we can safely import
+from pipecat.audio.vad.krisp_viva_vad import KrispVivaVadAnalyzer
+from pipecat.audio.vad.vad_analyzer import VADParams
+
+
+class TestKrispVivaVadAnalyzer(unittest.TestCase):
+    """Test suite for KrispVivaVadAnalyzer."""
+
+    def setUp(self):
+        """Set up test fixtures before each test method."""
+        # Create a temporary .kef model file for testing
+        self.temp_model_file = tempfile.NamedTemporaryFile(suffix=".kef", delete=False)
+        self.temp_model_file.write(b"dummy model data")
+        self.temp_model_file.close()
+        self.model_path = self.temp_model_file.name
+
+        # Use the global mock_krisp_audio that was set up before imports
+        self.mock_krisp_audio = mock_krisp_audio
+
+        # Reset all mocks to clear call counts from previous tests
+        self.mock_krisp_audio.reset_mock()
+        self.mock_krisp_audio.ModelInfo.reset_mock()
+        self.mock_krisp_audio.VadSessionConfig.reset_mock()
+        self.mock_krisp_audio.VadFloat.reset_mock()
+
+        # Mock ModelInfo
+        self.mock_model_info = MagicMock()
+        self.mock_krisp_audio.ModelInfo.return_value = self.mock_model_info
+
+        # Mock VadSessionConfig
+        self.mock_vad_cfg = MagicMock()
+        self.mock_krisp_audio.VadSessionConfig.return_value = self.mock_vad_cfg
+
+        # Mock VAD session
+        self.mock_session = MagicMock()
+        self.mock_session.process = MagicMock(return_value=0.75)  # Return voice probability
+        self.mock_krisp_audio.VadFloat.create.return_value = self.mock_session
+
+        # Patch krisp_audio in the module
+        self.krisp_audio_patch = patch(
+            "pipecat.audio.vad.krisp_viva_vad.krisp_audio", self.mock_krisp_audio
+        )
+        self.krisp_audio_patch.start()
+
+        # Patch KrispVivaSDKManager
+        self.sdk_manager_patcher = patch("pipecat.audio.vad.krisp_viva_vad.KrispVivaSDKManager")
+        self.mock_sdk_manager = self.sdk_manager_patcher.start()
+        self.mock_sdk_manager.acquire = MagicMock()
+        self.mock_sdk_manager.release = MagicMock()
+
+    def tearDown(self):
+        """Clean up test fixtures after each test method."""
+        # Stop all patchers
+        self.krisp_audio_patch.stop()
+        self.sdk_manager_patcher.stop()
+
+        # Remove temporary model file
+        if os.path.exists(self.model_path):
+            os.unlink(self.model_path)
+
+    def test_initialization_with_model_path(self):
+        """Test analyzer initialization with explicit model path."""
+        analyzer = KrispVivaVadAnalyzer(model_path=self.model_path)
+
+        # Verify SDK was acquired during initialization
+        self.mock_sdk_manager.acquire.assert_called_once()
+
+        # Verify analyzer attributes
+        self.assertEqual(analyzer._model_path, self.model_path)
+        self.assertEqual(analyzer._frame_duration_ms, 10)  # Default frame duration
+        self.assertIsNone(analyzer._session)  # Session created in set_sample_rate
+
+    def test_initialization_with_env_variable(self):
+        """Test analyzer initialization using KRISP_VIVA_VAD_MODEL_PATH environment variable."""
+        with patch.dict(os.environ, {"KRISP_VIVA_VAD_MODEL_PATH": self.model_path}):
+            analyzer = KrispVivaVadAnalyzer()
+
+            self.mock_sdk_manager.acquire.assert_called_once()
+            self.assertEqual(analyzer._model_path, self.model_path)
+
+    def test_initialization_without_model_path(self):
+        """Test analyzer initialization fails without model path."""
+        with patch.dict(os.environ, {}, clear=True):
+            with self.assertRaises(ValueError) as context:
+                KrispVivaVadAnalyzer()
+
+            self.assertIn("Model path", str(context.exception))
+            # acquire() is not called because exception is raised before it
+            self.mock_sdk_manager.acquire.assert_not_called()
+            # release() is called in the initialization exception handler
+            self.assertGreaterEqual(self.mock_sdk_manager.release.call_count, 1)
+
+    def test_initialization_with_invalid_extension(self):
+        """Test analyzer initialization fails with non-.kef file."""
+        with tempfile.NamedTemporaryFile(suffix=".txt", delete=False) as tmp:
+            tmp.write(b"dummy")
+            tmp_path = tmp.name
+
+        try:
+            with self.assertRaises(Exception) as context:
+                KrispVivaVadAnalyzer(model_path=tmp_path)
+
+            self.assertIn(".kef extension", str(context.exception))
+            # acquire() is not called because exception is raised before it
+            self.mock_sdk_manager.acquire.assert_not_called()
+            # release() is called in the initialization exception handler
+            self.assertGreaterEqual(self.mock_sdk_manager.release.call_count, 1)
+        finally:
+            os.unlink(tmp_path)
+
+    def test_initialization_with_nonexistent_file(self):
+        """Test analyzer initialization fails with non-existent model file."""
+        with self.assertRaises(FileNotFoundError):
+            KrispVivaVadAnalyzer(model_path="/nonexistent/path/model.kef")
+
+        # acquire() is not called because exception is raised before it
+        self.mock_sdk_manager.acquire.assert_not_called()
+        # release() is called in the initialization exception handler
+        self.assertGreaterEqual(self.mock_sdk_manager.release.call_count, 1)
+
+    def test_initialization_with_custom_frame_duration(self):
+        """Test analyzer initialization with custom frame duration."""
+        analyzer = KrispVivaVadAnalyzer(model_path=self.model_path, frame_duration=20)
+
+        self.assertEqual(analyzer._frame_duration_ms, 20)
+
+    def test_initialization_with_sample_rate(self):
+        """Test analyzer initialization with sample rate."""
+        analyzer = KrispVivaVadAnalyzer(
+            model_path=self.model_path, sample_rate=16000, frame_duration=10
+        )
+
+        # Should calculate samples per frame
+        self.assertEqual(analyzer._samples_per_frame, 160)  # 16000 * 10 / 1000
+
+    def test_set_sample_rate_with_supported_rates(self):
+        """Test setting sample rate with supported values."""
+        analyzer = KrispVivaVadAnalyzer(model_path=self.model_path)
+
+        for sample_rate in [8000, 16000, 32000, 44100, 48000]:
+            analyzer.set_sample_rate(sample_rate)
+
+            # Verify session was created
+            self.assertIsNotNone(analyzer._session)
+            self.assertEqual(analyzer.sample_rate, sample_rate)
+
+            # Verify samples per frame was calculated
+            expected_samples = int((sample_rate * analyzer._frame_duration_ms) / 1000)
+            self.assertEqual(analyzer._samples_per_frame, expected_samples)
+
+            # Verify VadSessionConfig was created and configured
+            self.mock_krisp_audio.VadSessionConfig.assert_called()
+            self.assertEqual(self.mock_vad_cfg.modelInfo, self.mock_model_info)
+
+    def test_set_sample_rate_with_unsupported_rate(self):
+        """Test setting sample rate with unsupported value raises ValueError."""
+        analyzer = KrispVivaVadAnalyzer(model_path=self.model_path)
+
+        with self.assertRaises(ValueError) as context:
+            analyzer.set_sample_rate(12000)  # Unsupported sample rate
+
+        self.assertIn("sample rate needs to be", str(context.exception))
+
+    def test_num_frames_required_with_session(self):
+        """Test num_frames_required when session is created."""
+        analyzer = KrispVivaVadAnalyzer(
+            model_path=self.model_path, sample_rate=16000, frame_duration=10
+        )
+        analyzer.set_sample_rate(16000)
+
+        # Should return samples per frame from session
+        frames_required = analyzer.num_frames_required()
+        self.assertEqual(frames_required, 160)  # 16000 * 10 / 1000
+
+    def test_num_frames_required_without_session(self):
+        """Test num_frames_required when session is not created yet."""
+        analyzer = KrispVivaVadAnalyzer(
+            model_path=self.model_path, sample_rate=16000, frame_duration=10
+        )
+
+        # Should calculate from sample_rate
+        frames_required = analyzer.num_frames_required()
+        self.assertEqual(frames_required, 160)
+
+    def test_num_frames_required_with_different_sample_rates(self):
+        """Test num_frames_required with different sample rates."""
+        test_cases = [
+            (8000, 10, 80),  # 8kHz @ 10ms = 80 samples
+            (16000, 10, 160),  # 16kHz @ 10ms = 160 samples
+            (16000, 20, 320),  # 16kHz @ 20ms = 320 samples
+            (32000, 10, 320),  # 32kHz @ 10ms = 320 samples
+            (44100, 10, 441),  # 44.1kHz @ 10ms = 441 samples
+            (48000, 10, 480),  # 48kHz @ 10ms = 480 samples
+        ]
+
+        for sample_rate, frame_duration, expected_frames in test_cases:
+            analyzer = KrispVivaVadAnalyzer(
+                model_path=self.model_path,
+                sample_rate=sample_rate,
+                frame_duration=frame_duration,
+            )
+            analyzer.set_sample_rate(sample_rate)
+
+            frames_required = analyzer.num_frames_required()
+            self.assertEqual(
+                frames_required,
+                expected_frames,
+                f"Failed for {sample_rate}Hz @ {frame_duration}ms",
+            )
+
+    def test_num_frames_required_fallback(self):
+        """Test num_frames_required fallback when sample rate not set."""
+        analyzer = KrispVivaVadAnalyzer(model_path=self.model_path, frame_duration=10)
+
+        # Should use default fallback (16kHz)
+        frames_required = analyzer.num_frames_required()
+        self.assertEqual(frames_required, 160)  # 16000 * 10 / 1000
+
+    def test_voice_confidence_with_valid_buffer(self):
+        """Test voice_confidence with valid audio buffer."""
+        analyzer = KrispVivaVadAnalyzer(model_path=self.model_path)
+        analyzer.set_sample_rate(16000)
+
+        # Create audio buffer for one frame (160 samples = 320 bytes)
+        samples = np.random.randint(-32768, 32767, size=160, dtype=np.int16)
+        audio_buffer = samples.tobytes()
+
+        confidence = analyzer.voice_confidence(audio_buffer)
+
+        # Verify confidence is returned
+        self.assertIsInstance(confidence, float)
+        self.assertEqual(confidence, 0.75)  # Mock returns 0.75
+
+        # Verify session.process was called
+        self.mock_session.process.assert_called_once()
+
+    def test_voice_confidence_without_session(self):
+        """Test voice_confidence when session is not initialized."""
+        analyzer = KrispVivaVadAnalyzer(model_path=self.model_path)
+
+        # Create audio buffer
+        samples = np.random.randint(-32768, 32767, size=160, dtype=np.int16)
+        audio_buffer = samples.tobytes()
+
+        # Should return 0.0 and log warning
+        confidence = analyzer.voice_confidence(audio_buffer)
+        self.assertEqual(confidence, 0.0)
+
+        # Verify session.process was NOT called
+        self.mock_session.process.assert_not_called()
+
+    def test_voice_confidence_error_handling(self):
+        """Test voice_confidence handles processing errors gracefully."""
+        analyzer = KrispVivaVadAnalyzer(model_path=self.model_path)
+        analyzer.set_sample_rate(16000)
+
+        # Make session.process raise an exception
+        self.mock_session.process.side_effect = Exception("Processing error")
+
+        # Create audio buffer
+        samples = np.random.randint(-32768, 32767, size=160, dtype=np.int16)
+        audio_buffer = samples.tobytes()
+
+        # Should return 0.0 on error
+        confidence = analyzer.voice_confidence(audio_buffer)
+        self.assertEqual(confidence, 0.0)
+
+    def test_voice_confidence_audio_conversion(self):
+        """Test voice_confidence properly converts audio format."""
+        analyzer = KrispVivaVadAnalyzer(model_path=self.model_path)
+        analyzer.set_sample_rate(16000)
+
+        # Create deterministic audio buffer
+        samples = np.array([1000, -2000, 3000, -4000], dtype=np.int16)
+        audio_buffer = samples.tobytes()
+
+        analyzer.voice_confidence(audio_buffer)
+
+        # Verify process was called with float32 array
+        call_args = self.mock_session.process.call_args[0][0]
+        self.assertIsInstance(call_args, np.ndarray)
+        self.assertEqual(call_args.dtype, np.float32)
+
+        # Verify normalization (int16 to float32)
+        expected_float = samples.astype(np.float32) / 32768.0
+        np.testing.assert_array_almost_equal(call_args, expected_float)
+
+    def test_voice_confidence_different_buffer_sizes(self):
+        """Test voice_confidence with different buffer sizes."""
+        analyzer = KrispVivaVadAnalyzer(model_path=self.model_path)
+        analyzer.set_sample_rate(16000)
+
+        test_sizes = [80, 160, 320, 480]  # Different numbers of samples
+
+        for size in test_sizes:
+            samples = np.random.randint(-32768, 32767, size=size, dtype=np.int16)
+            audio_buffer = samples.tobytes()
+
+            confidence = analyzer.voice_confidence(audio_buffer)
+
+            # Should always return a float
+            self.assertIsInstance(confidence, float)
+            self.assertGreaterEqual(confidence, 0.0)
+            self.assertLessEqual(confidence, 1.0)
+
+    def test_cleanup(self):
+        """Test that cleanup releases resources explicitly."""
+        analyzer = KrispVivaVadAnalyzer(model_path=self.model_path)
+        analyzer.set_sample_rate(16000)
+
+        # Reset mock to track calls
+        self.mock_sdk_manager.release.reset_mock()
+
+        asyncio.run(analyzer.cleanup())
+
+        # Verify SDK was released and session was cleared
+        self.mock_sdk_manager.release.assert_called_once()
+        self.assertIsNone(analyzer._session)
+
+    def test_initialization_with_vad_params(self):
+        """Test analyzer initialization with VAD parameters."""
+        params = VADParams(confidence=0.8, start_secs=0.3, stop_secs=0.9)
+        analyzer = KrispVivaVadAnalyzer(
+            model_path=self.model_path, sample_rate=16000, params=params
+        )
+
+        self.assertEqual(analyzer.params.confidence, 0.8)
+        self.assertEqual(analyzer.params.start_secs, 0.3)
+        self.assertEqual(analyzer.params.stop_secs, 0.9)
+
+    def test_set_sample_rate_creates_session(self):
+        """Test that set_sample_rate creates a new session."""
+        analyzer = KrispVivaVadAnalyzer(model_path=self.model_path)
+
+        # Initially no session
+        self.assertIsNone(analyzer._session)
+
+        # Set sample rate should create session
+        analyzer.set_sample_rate(16000)
+
+        # Verify session was created
+        self.assertIsNotNone(analyzer._session)
+        self.mock_krisp_audio.VadFloat.create.assert_called_once()
+
+    def test_set_sample_rate_replaces_session(self):
+        """Test that set_sample_rate replaces existing session."""
+        analyzer = KrispVivaVadAnalyzer(model_path=self.model_path)
+
+        # Create first session
+        analyzer.set_sample_rate(16000)
+        session1 = analyzer._session
+        self.assertIsNotNone(session1)
+
+        # Reset mock to track new calls
+        self.mock_krisp_audio.VadFloat.create.reset_mock()
+
+        # Set different sample rate should create new session
+        analyzer.set_sample_rate(48000)
+        session2 = analyzer._session
+        self.assertIsNotNone(session2)
+
+        # Verify new session was created
+        self.mock_krisp_audio.VadFloat.create.assert_called_once()
+
+    def test_session_configuration(self):
+        """Test that session is configured correctly."""
+        analyzer = KrispVivaVadAnalyzer(model_path=self.model_path, frame_duration=15)
+        analyzer.set_sample_rate(16000)
+
+        # Verify VadSessionConfig was configured
+        self.assertEqual(self.mock_vad_cfg.modelInfo, self.mock_model_info)
+        self.assertEqual(self.mock_vad_cfg.inputFrameDuration, "15ms")
+
+        # Verify sample rate was set
+        from pipecat.audio.krisp_instance import int_to_krisp_sample_rate
+
+        expected_sample_rate = int_to_krisp_sample_rate(16000)
+        self.assertEqual(self.mock_vad_cfg.inputSampleRate, expected_sample_rate)
+
+    def test_multiple_sample_rate_changes(self):
+        """Test multiple sample rate changes."""
+        analyzer = KrispVivaVadAnalyzer(model_path=self.model_path)
+
+        sample_rates = [8000, 16000, 32000, 48000]
+
+        for sample_rate in sample_rates:
+            analyzer.set_sample_rate(sample_rate)
+            self.assertEqual(analyzer.sample_rate, sample_rate)
+
+            # Verify num_frames_required is correct
+            expected_frames = int((sample_rate * 10) / 1000)
+            self.assertEqual(analyzer.num_frames_required(), expected_frames)
+
+
+if __name__ == "__main__":
+    unittest.main()