diff --git a/src/pipecat/turns/user_start/__init__.py b/src/pipecat/turns/user_start/__init__.py index 14de5d28b..c2b1efa73 100644 --- a/src/pipecat/turns/user_start/__init__.py +++ b/src/pipecat/turns/user_start/__init__.py @@ -14,7 +14,7 @@ from .wake_phrase_user_turn_start_strategy import WakePhraseUserTurnStartStrateg try: from .krisp_viva_ip_user_turn_start_strategy import KrispVivaIPUserTurnStartStrategy except ImportError: - KrispVivaIPUserTurnStartStrategy = None + KrispVivaIPUserTurnStartStrategy = None # krisp_audio not installed __all__ = [ "BaseUserTurnStartStrategy", diff --git a/src/pipecat/turns/user_start/krisp_viva_ip_user_turn_start_strategy.py b/src/pipecat/turns/user_start/krisp_viva_ip_user_turn_start_strategy.py index 807bc8154..a366b8847 100644 --- a/src/pipecat/turns/user_start/krisp_viva_ip_user_turn_start_strategy.py +++ b/src/pipecat/turns/user_start/krisp_viva_ip_user_turn_start_strategy.py @@ -43,7 +43,7 @@ except ModuleNotFoundError as e: logger.error( "In order to use KrispVivaIPUserTurnStartStrategy, you need to install krisp_audio." ) - raise Exception(f"Missing module: {e}") + raise ImportError(f"Missing module: {e}") from e class KrispVivaIPUserTurnStartStrategy(BaseUserTurnStartStrategy): @@ -211,22 +211,24 @@ class KrispVivaIPUserTurnStartStrategy(BaseUserTurnStartStrategy): """ logger.trace("Krisp VIVA IP: VAD speech started, collecting audio for classification") self._speech_active = True - self._audio_buffer.clear() self._decision_made = False return ProcessFrameResult.CONTINUE async def _handle_audio(self, frame: InputAudioRawFrame) -> ProcessFrameResult: """Feed audio to the IP model and check for genuine interruption. + Every frame is passed to the IP model regardless of speech state so + that the model maintains continuous internal state (matching the + standalone Krisp SDK behaviour). ``_speech_active`` is forwarded as + the per-frame VAD input to the model and also gates the threshold + evaluation. + Args: frame: Raw audio input frame. Returns: STOP if the model detects a genuine interruption, CONTINUE otherwise. """ - if not self._speech_active or self._decision_made: - return ProcessFrameResult.CONTINUE - self._ensure_session(frame.sample_rate) if self._ip_session is None or self._samples_per_frame is None: @@ -254,7 +256,7 @@ class KrispVivaIPUserTurnStartStrategy(BaseUserTurnStartStrategy): for ip_frame in frames: ip_prob = self._ip_session.process(ip_frame.tolist(), self._speech_active) - if ip_prob >= self._threshold: + if self._speech_active and not self._decision_made and ip_prob >= self._threshold: logger.debug( f"Krisp VIVA IP: genuine interruption detected (prob={ip_prob:.3f}, " f"threshold={self._threshold})" diff --git a/tests/test_krisp_ip_user_turn_start_strategy.py b/tests/test_krisp_ip_user_turn_start_strategy.py index bb34d879a..5e5dd00cc 100644 --- a/tests/test_krisp_ip_user_turn_start_strategy.py +++ b/tests/test_krisp_ip_user_turn_start_strategy.py @@ -103,8 +103,12 @@ class TestKrispVivaIPUserTurnStartStrategy(unittest.IsolatedAsyncioTestCase): api_key="test-key", ) - def _audio_frame(self, sample_rate: int = 16000, frame_duration_ms: int = 20): - samples = int(sample_rate * frame_duration_ms / 1000) + def _audio_frame( + self, sample_rate: int = 16000, frame_duration_ms: int = 20, num_samples: int | None = None + ): + samples = ( + num_samples if num_samples is not None else int(sample_rate * frame_duration_ms / 1000) + ) return InputAudioRawFrame( audio=_int16_silence(samples), sample_rate=sample_rate, @@ -165,7 +169,10 @@ class TestKrispVivaIPUserTurnStartStrategy(unittest.IsolatedAsyncioTestCase): result = await strategy.process_frame(self._audio_frame()) self.assertEqual(result, ProcessFrameResult.CONTINUE) - self.mock_ip_session.process.assert_not_called() + # process() is still called (continuous state), but with speech_active=False + self.mock_ip_session.process.assert_called() + args = self.mock_ip_session.process.call_args[0] + self.assertFalse(args[1]) # speech_active should be False finally: await strategy.cleanup() @@ -182,7 +189,10 @@ class TestKrispVivaIPUserTurnStartStrategy(unittest.IsolatedAsyncioTestCase): result = await strategy.process_frame(self._audio_frame()) self.assertEqual(result, ProcessFrameResult.CONTINUE) - self.mock_ip_session.process.assert_not_called() + # process() is still called (continuous state), but with speech_active=False + self.mock_ip_session.process.assert_called() + args = self.mock_ip_session.process.call_args[0] + self.assertFalse(args[1]) # speech_active should be False finally: await strategy.cleanup() @@ -193,7 +203,11 @@ class TestKrispVivaIPUserTurnStartStrategy(unittest.IsolatedAsyncioTestCase): try: result = await strategy.process_frame(self._audio_frame()) self.assertEqual(result, ProcessFrameResult.CONTINUE) - self.mock_ip_session.process.assert_not_called() + # process() is called (continuous state) even before VAD start, + # but _speech_active=False prevents triggering despite high prob + self.mock_ip_session.process.assert_called() + args = self.mock_ip_session.process.call_args[0] + self.assertFalse(args[1]) # speech_active should be False finally: await strategy.cleanup() @@ -219,6 +233,21 @@ class TestKrispVivaIPUserTurnStartStrategy(unittest.IsolatedAsyncioTestCase): finally: await strategy.cleanup() + async def test_reset_clears_audio_buffer(self): + self.mock_ip_session.process = MagicMock(return_value=0.1) + + strategy = self._make_strategy(threshold=0.5) + try: + await strategy.process_frame(VADUserStartedSpeakingFrame()) + # Feed a partial frame (smaller than samples_per_frame) so it stays in buffer + await strategy.process_frame(self._audio_frame(num_samples=10)) + self.assertGreater(len(strategy._audio_buffer), 0) + + await strategy.process_frame(VADUserStoppedSpeakingFrame()) + self.assertEqual(len(strategy._audio_buffer), 0) + finally: + await strategy.cleanup() + async def test_unrelated_frames_continue(self): strategy = self._make_strategy() try: @@ -231,6 +260,87 @@ class TestKrispVivaIPUserTurnStartStrategy(unittest.IsolatedAsyncioTestCase): finally: await strategy.cleanup() + async def test_reset_method_clears_state(self): + self.mock_ip_session.process = MagicMock(return_value=0.1) + + strategy = self._make_strategy(threshold=0.5) + try: + await strategy.process_frame(VADUserStartedSpeakingFrame()) + await strategy.process_frame(self._audio_frame(num_samples=10)) + self.assertTrue(strategy._speech_active) + self.assertGreater(len(strategy._audio_buffer), 0) + + await strategy.reset() + + self.assertFalse(strategy._speech_active) + self.assertEqual(len(strategy._audio_buffer), 0) + self.assertFalse(strategy._decision_made) + finally: + await strategy.cleanup() + + async def test_cleanup_releases_sdk(self): + strategy = self._make_strategy() + + await strategy.cleanup() + + self.mock_sdk_manager.release.assert_called_once() + self.assertIsNone(strategy._ip_session) + self.assertFalse(strategy._sdk_acquired) + + def test_init_raises_if_no_model_path(self): + with self.assertRaises(ValueError): + KrispVivaIPUserTurnStartStrategy(api_key="test-key") + + def test_init_raises_if_wrong_extension(self): + with self.assertRaises(ValueError): + KrispVivaIPUserTurnStartStrategy(model_path="/some/model.bin", api_key="test-key") + + def test_init_raises_if_file_not_found(self): + with self.assertRaises(FileNotFoundError): + KrispVivaIPUserTurnStartStrategy( + model_path="/nonexistent/model.kef", api_key="test-key" + ) + + def test_init_raises_if_sdk_fails(self): + self.mock_sdk_manager.acquire.side_effect = RuntimeError("SDK error") + with self.assertRaises(RuntimeError): + KrispVivaIPUserTurnStartStrategy(model_path=self.model_path, api_key="test-key") + + def test_init_uses_env_var_for_model_path(self): + with patch.dict(os.environ, {"KRISP_VIVA_IP_MODEL_PATH": self.model_path}): + strategy = KrispVivaIPUserTurnStartStrategy(api_key="test-key") + self.assertEqual(strategy._model_path, self.model_path) + + async def test_vad_stopped_when_speech_inactive_is_no_op(self): + strategy = self._make_strategy() + try: + result = await strategy.process_frame(VADUserStoppedSpeakingFrame()) + self.assertEqual(result, ProcessFrameResult.CONTINUE) + self.assertFalse(strategy._speech_active) + finally: + await strategy.cleanup() + + async def test_interruption_at_exact_threshold_triggers(self): + threshold = 0.5 + self.mock_ip_session.process = MagicMock(return_value=threshold) + + strategy = self._make_strategy(threshold=threshold) + try: + fired = False + + @strategy.event_handler("on_user_turn_started") + async def on_user_turn_started(strategy, params): + nonlocal fired + fired = True + + await strategy.process_frame(VADUserStartedSpeakingFrame()) + result = await strategy.process_frame(self._audio_frame()) + + self.assertTrue(fired) + self.assertEqual(result, ProcessFrameResult.STOP) + finally: + await strategy.cleanup() + if __name__ == "__main__": unittest.main()