Merge pull request #3809 from pipecat-ai/mb/krisp-viva-result

Add Krisp API key support and debug logging

changelog/3809.added.md

@@ -0,0 +1 @@
+- Added `TurnMetricsData` as a generic metrics class for turn detection, with e2e processing time measurement. `KrispVivaTurn` now emits `TurnMetricsData` with `e2e_processing_time_ms` tracking the interval from VAD speech-to-silence transition to turn completion.

changelog/3809.changed.md

@@ -0,0 +1 @@
+- Added `api_key` parameter to `KrispVivaSDKManager`, `KrispVivaTurn`, and `KrispVivaFilter` for Krisp SDK v1.6.1+ licensing. Falls back to `KRISP_VIVA_API_KEY` environment variable.

changelog/3809.deprecated.md

@@ -0,0 +1 @@
+- Deprecated `SmartTurnMetricsData` in favor of `TurnMetricsData`. `BaseSmartTurn` now emits `TurnMetricsData` directly.

env.example

@@ -104,6 +104,7 @@ INWORLD_API_KEY=...
 KRISP_MODEL_PATH=...
 
 # Krisp Viva
+KRISP_VIVA_API_KEY=...
 KRISP_VIVA_FILTER_MODEL_PATH=...
 KRISP_VIVA_TURN_MODEL_PATH=...
 

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

@@ -31,6 +31,8 @@ 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.frames.frames import LLMRunFrame
+from pipecat.metrics.metrics import TurnMetricsData
+from pipecat.observers.loggers.metrics_log_observer import MetricsLogObserver
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
@@ -41,32 +43,37 @@ from pipecat.processors.aggregators.llm_response_universal import (
 )
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
+from pipecat.services.cartesia.tts import CartesiaTTSService
 from pipecat.services.deepgram.stt import DeepgramSTTService
-from pipecat.services.deepgram.tts import DeepgramTTSService
 from pipecat.services.openai.llm import OpenAILLMService
 from pipecat.transports.base_transport import BaseTransport, TransportParams
 from pipecat.transports.daily.transport import DailyParams
 from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+from pipecat.turns.user_stop import TurnAnalyzerUserTurnStopStrategy
+from pipecat.turns.user_turn_strategies import UserTurnStrategies
 
 load_dotenv(override=True)
 
 # We use lambdas to defer transport parameter creation until the transport
 # type is selected at runtime.
+
+krisp_viva_filter = KrispVivaFilter()
+
 transport_params = {
     "daily": lambda: DailyParams(
         audio_in_enabled=True,
         audio_out_enabled=True,
-        audio_in_filter=KrispVivaFilter(),
+        audio_in_filter=krisp_viva_filter,
     ),
     "twilio": lambda: FastAPIWebsocketParams(
         audio_in_enabled=True,
         audio_out_enabled=True,
-        audio_in_filter=KrispVivaFilter(),
+        audio_in_filter=krisp_viva_filter,
     ),
     "webrtc": lambda: TransportParams(
         audio_in_enabled=True,
         audio_out_enabled=True,
-        audio_in_filter=KrispVivaFilter(),
+        audio_in_filter=krisp_viva_filter,
     ),
 }
 
@@ -76,7 +83,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
 
     stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
 
-    tts = DeepgramTTSService(api_key=os.getenv("DEEPGRAM_API_KEY"), voice="aura-helios-en")
+    tts = CartesiaTTSService(
+        api_key=os.getenv("CARTESIA_API_KEY"), voice_id="71a7ad14-091c-4e8e-a314-022ece01c121"
+    )
 
     llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
 
@@ -117,6 +126,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
             enable_usage_metrics=True,
         ),
         idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
+        observers=[MetricsLogObserver(include_metrics={TurnMetricsData})],
     )
 
     @transport.event_handler("on_client_connected")

examples/foundational/38b-smart-turn-local.py

@@ -12,6 +12,8 @@ from loguru import logger
 
 from pipecat.audio.vad.silero import SileroVADAnalyzer
 from pipecat.frames.frames import LLMRunFrame
+from pipecat.metrics.metrics import TurnMetricsData
+from pipecat.observers.loggers.metrics_log_observer import MetricsLogObserver
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
@@ -77,7 +79,6 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     pipeline = Pipeline(
         [
             transport.input(),  # Transport user input
-            rtvi,
             stt,
             user_aggregator,  # User responses
             llm,  # LLM
@@ -94,17 +95,15 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
             enable_usage_metrics=True,
         ),
         idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
+        observers=[MetricsLogObserver(include_metrics={TurnMetricsData})],
     )
 
-    @task.rtvi.event_handler("on_client_ready")
-    async def on_client_ready(rtvi):
-        # Kick off the conversation
-        messages.append({"role": "system", "content": "Please introduce yourself to the user."})
-        await task.queue_frames([LLMRunFrame()])
-
     @transport.event_handler("on_client_connected")
     async def on_client_connected(transport, client):
         logger.info(f"Client connected")
+        # Kick off the conversation
+        messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+        await task.queue_frames([LLMRunFrame()])
 
     @transport.event_handler("on_client_disconnected")
     async def on_client_disconnected(transport, client):

scripts/evals/run-release-evals.py

@@ -123,6 +123,7 @@ TESTS_07 = [
     ("07n-interruptible-google.py", EVAL_SIMPLE_MATH),
     ("07n-interruptible-google-http.py", EVAL_SIMPLE_MATH),
     ("07o-interruptible-assemblyai.py", EVAL_SIMPLE_MATH),
+    ("07p-interruptible-krisp-viva.py", EVAL_SIMPLE_MATH),
     ("07q-interruptible-rime.py", EVAL_SIMPLE_MATH),
     ("07q-interruptible-rime-http.py", EVAL_SIMPLE_MATH),
     ("07r-interruptible-nvidia.py", EVAL_SIMPLE_MATH),
@@ -148,8 +149,6 @@ TESTS_07 = [
     ("07zj-interruptible-kokoro.py", EVAL_SIMPLE_MATH),
     # Needs a local XTTS docker instance running.
     # ("07i-interruptible-xtts.py", EVAL_SIMPLE_MATH),
-    # Needs a Krisp license.
-    # ("07p-interruptible-krisp.py", EVAL_SIMPLE_MATH),
 ]
 
 TESTS_12 = [

src/pipecat/audio/filters/krisp_viva_filter.py

@@ -39,7 +39,11 @@ class KrispVivaFilter(BaseAudioFilter):
     """
 
     def __init__(
-        self, model_path: str = None, frame_duration: int = 10, noise_suppression_level: int = 100
+        self,
+        model_path: str = None,
+        frame_duration: int = 10,
+        noise_suppression_level: int = 100,
+        api_key: str = "",
     ) -> None:
         """Initialize the Krisp noise reduction filter.
 
@@ -48,6 +52,8 @@ class KrispVivaFilter(BaseAudioFilter):
                 If None, uses KRISP_VIVA_FILTER_MODEL_PATH environment variable.
             frame_duration: Frame duration in milliseconds.
             noise_suppression_level: Noise suppression level.
+            api_key: Krisp SDK API key. If empty, falls back to
+                the KRISP_VIVA_API_KEY environment variable.
 
         Raises:
             ValueError: If model_path is not provided and KRISP_VIVA_FILTER_MODEL_PATH is not set.
@@ -57,6 +63,8 @@ class KrispVivaFilter(BaseAudioFilter):
         """
         super().__init__()
 
+        self._api_key = api_key
+
         try:
             # Set model path, checking environment if not specified
             if model_path:
@@ -132,7 +140,7 @@ class KrispVivaFilter(BaseAudioFilter):
         """
         try:
             # Acquire SDK reference (will initialize on first call)
-            KrispVivaSDKManager.acquire()
+            KrispVivaSDKManager.acquire(api_key=self._api_key)
             self._session = self._create_session(sample_rate, self._frame_duration_ms)
         except Exception as e:
             logger.error(f"Failed to start Krisp session: {e}", exc_info=True)

src/pipecat/audio/krisp_instance.py

@@ -7,6 +7,7 @@
 """Krisp Instance manager for pipecat audio."""
 
 import atexit
+import os
 from threading import Lock
 
 from loguru import logger
@@ -88,17 +89,26 @@ class KrispVivaSDKManager:
     _lock = Lock()
     _reference_count = 0
 
+    @staticmethod
+    def _license_callback(error, error_message):
+        """Callback for Krisp SDK licensing errors."""
+        logger.error(f"Krisp licensing error: {error} - {error_message}")
+
     @staticmethod
     def _log_callback(log_message, log_level):
         """Thread-safe callback for Krisp SDK logging."""
         logger.info(f"[{log_level}] {log_message}")
 
     @classmethod
-    def acquire(cls):
+    def acquire(cls, api_key: str = ""):
         """Acquire a reference to the SDK (initializes if needed).
 
         Call this when creating a filter instance.
 
+        Args:
+            api_key: Krisp SDK API key. If empty, falls back to the
+                KRISP_VIVA_API_KEY environment variable.
+
         Raises:
             Exception: If SDK initialization fails (propagated from krisp_audio)
         """
@@ -106,7 +116,19 @@ class KrispVivaSDKManager:
             # Initialize SDK on first acquire
             if cls._reference_count == 0:
                 try:
-                    krisp_audio.globalInit("", cls._log_callback, krisp_audio.LogLevel.Off)
+                    key = api_key or os.environ.get("KRISP_VIVA_API_KEY", "")
+                    try:
+                        # New SDK signature (requires license key)
+                        krisp_audio.globalInit(
+                            "",
+                            key,
+                            cls._license_callback,
+                            cls._log_callback,
+                            krisp_audio.LogLevel.Off,
+                        )
+                    except TypeError:
+                        # Old SDK signature (no license key)
+                        krisp_audio.globalInit("", cls._log_callback, krisp_audio.LogLevel.Off)
 
                     cls._initialized = True
 

src/pipecat/audio/turn/krisp_viva_turn.py

@@ -15,6 +15,7 @@ passed directly to the constructor.
 """
 
 import os
+import time
 from typing import Optional, Tuple
 
 import numpy as np
@@ -26,7 +27,7 @@ from pipecat.audio.krisp_instance import (
     int_to_krisp_sample_rate,
 )
 from pipecat.audio.turn.base_turn_analyzer import BaseTurnAnalyzer, BaseTurnParams, EndOfTurnState
-from pipecat.metrics.metrics import MetricsData
+from pipecat.metrics.metrics import MetricsData, TurnMetricsData
 
 try:
     import krisp_audio
@@ -63,6 +64,7 @@ class KrispVivaTurn(BaseTurnAnalyzer):
         model_path: Optional[str] = None,
         sample_rate: Optional[int] = None,
         params: Optional[KrispTurnParams] = None,
+        api_key: str = "",
     ) -> None:
         """Initialize the Krisp turn analyzer.
 
@@ -72,6 +74,8 @@ class KrispVivaTurn(BaseTurnAnalyzer):
             sample_rate: Optional initial sample rate for audio processing.
                 If provided, this will be used as the fixed sample rate.
             params: Configuration parameters for turn analysis behavior.
+            api_key: Krisp SDK API key. If empty, falls back to
+                the KRISP_VIVA_API_KEY environment variable.
 
         Raises:
             ValueError: If model_path is not provided and KRISP_VIVA_TURN_MODEL_PATH is not set.
@@ -83,7 +87,7 @@ class KrispVivaTurn(BaseTurnAnalyzer):
 
         # Acquire SDK reference (will initialize on first call)
         try:
-            KrispVivaSDKManager.acquire()
+            KrispVivaSDKManager.acquire(api_key=api_key)
             self._sdk_acquired = True
         except Exception as e:
             self._sdk_acquired = False
@@ -115,6 +119,9 @@ class KrispVivaTurn(BaseTurnAnalyzer):
             self._last_probability = None
             self._frame_probabilities = []
             self._last_state = EndOfTurnState.INCOMPLETE
+            self._speech_stopped_time: Optional[float] = None
+            self._e2e_processing_time_ms: Optional[float] = None
+            self._last_metrics: Optional[TurnMetricsData] = None
 
             # Create session with provided sample rate or default to 16000 Hz
             # This preloads the model to improve latency when set_sample_rate is called later
@@ -288,7 +295,14 @@ class KrispVivaTurn(BaseTurnAnalyzer):
                     # Track speech start time
                     if not self._speech_triggered:
                         logger.trace("Speech detected, turn analysis started")
+                        self._e2e_processing_time_ms = None
                     self._speech_triggered = True
+                    # Reset speech stopped time when speech resumes
+                    self._speech_stopped_time = None
+                else:
+                    # Record the moment speech transitions to non-speech
+                    if self._speech_triggered and self._speech_stopped_time is None:
+                        self._speech_stopped_time = time.perf_counter()
                 # Note: We don't immediately mark as complete on silence detection.
                 # Instead, we wait for the model's probability check below to confirm
                 # end-of-turn based on the threshold.
@@ -308,6 +322,18 @@ class KrispVivaTurn(BaseTurnAnalyzer):
                 # Only mark as complete if we've detected speech and the model
                 # confirms with sufficient confidence
                 if self._speech_triggered and prob >= self._params.threshold:
+                    # Calculate e2e processing time: time from speech stop to threshold crossing
+                    if self._speech_stopped_time is not None:
+                        self._e2e_processing_time_ms = (
+                            time.perf_counter() - self._speech_stopped_time
+                        ) * 1000
+                        self._last_metrics = TurnMetricsData(
+                            processor="KrispVivaTurn",
+                            is_complete=True,
+                            probability=prob,
+                            e2e_processing_time_ms=self._e2e_processing_time_ms,
+                        )
+                        logger.debug(f"Krisp turn complete")
                     state = EndOfTurnState.COMPLETE
                     self.clear()
                     break
@@ -329,12 +355,15 @@ class KrispVivaTurn(BaseTurnAnalyzer):
             Tuple containing the end-of-turn state and optional metrics data.
             Returns the last state determined by append_audio().
         """
-        # For real-time processing, the state is determined in append_audio
-        # Return the last state that was computed
-        return self._last_state, None
+        # For real-time processing, the state is determined in append_audio.
+        # Consume metrics so they aren't pushed twice.
+        metrics = self._last_metrics
+        self._last_metrics = None
+        return self._last_state, metrics
 
     def clear(self):
         """Reset the turn analyzer to its initial state."""
         self._speech_triggered = False
         self._audio_buffer.clear()
         self._last_state = EndOfTurnState.INCOMPLETE
+        self._speech_stopped_time = None

src/pipecat/audio/turn/smart_turn/base_smart_turn.py

@@ -21,7 +21,7 @@ import numpy as np
 from loguru import logger
 
 from pipecat.audio.turn.base_turn_analyzer import BaseTurnAnalyzer, BaseTurnParams, EndOfTurnState
-from pipecat.metrics.metrics import MetricsData, SmartTurnMetricsData
+from pipecat.metrics.metrics import MetricsData, TurnMetricsData
 
 # Default timing parameters
 STOP_SECS = 3
@@ -222,18 +222,11 @@ class BaseSmartTurn(BaseTurnAnalyzer):
                 # Calculate processing time
                 e2e_processing_time_ms = (end_time - start_time) * 1000
 
-                # Extract metrics from the nested structure
-                metrics = result.get("metrics", {})
-                inference_time = metrics.get("inference_time", 0)
-                total_time = metrics.get("total_time", 0)
-
                 # Prepare the result data
-                result_data = SmartTurnMetricsData(
+                result_data = TurnMetricsData(
                     processor="BaseSmartTurn",
                     is_complete=result["prediction"] == 1,
                     probability=result["probability"],
-                    inference_time_ms=inference_time * 1000,
-                    server_total_time_ms=total_time * 1000,
                     e2e_processing_time_ms=e2e_processing_time_ms,
                 )
 
@@ -241,8 +234,6 @@ class BaseSmartTurn(BaseTurnAnalyzer):
                     f"Prediction: {'Complete' if result_data.is_complete else 'Incomplete'}"
                 )
                 logger.trace(f"Probability of complete: {result_data.probability:.4f}")
-                logger.trace(f"Inference time: {result_data.inference_time_ms:.2f}ms")
-                logger.trace(f"Server total time: {result_data.server_total_time_ms:.2f}ms")
                 logger.trace(f"E2E processing time: {result_data.e2e_processing_time_ms:.2f}ms")
             except SmartTurnTimeoutException:
                 logger.debug(

src/pipecat/metrics/metrics.py

@@ -87,19 +87,31 @@ class TTSUsageMetricsData(MetricsData):
     value: int
 
 
-class SmartTurnMetricsData(MetricsData):
-    """Metrics data for smart turn predictions.
+class TurnMetricsData(MetricsData):
+    """Metrics data for turn detection predictions.
 
     Parameters:
         is_complete: Whether the turn is predicted to be complete.
         probability: Confidence probability of the turn completion prediction.
-        inference_time_ms: Time taken for inference in milliseconds.
-        server_total_time_ms: Total server processing time in milliseconds.
-        e2e_processing_time_ms: End-to-end processing time in milliseconds.
+        e2e_processing_time_ms: End-to-end processing time in milliseconds,
+            measured from VAD speech-to-silence transition to turn completion.
     """
 
     is_complete: bool
     probability: float
-    inference_time_ms: float
-    server_total_time_ms: float
     e2e_processing_time_ms: float
+
+
+class SmartTurnMetricsData(TurnMetricsData):
+    """Metrics data for smart turn predictions.
+
+    .. deprecated:: 0.0.104
+        Use :class:`TurnMetricsData` instead. This class will be removed in a future version.
+
+    Parameters:
+        inference_time_ms: Time taken for inference in milliseconds.
+        server_total_time_ms: Total server processing time in milliseconds.
+    """
+
+    inference_time_ms: float = 0.0
+    server_total_time_ms: float = 0.0

src/pipecat/observers/loggers/metrics_log_observer.py

@@ -24,6 +24,7 @@ from pipecat.metrics.metrics import (
     SmartTurnMetricsData,
     TTFBMetricsData,
     TTSUsageMetricsData,
+    TurnMetricsData,
 )
 from pipecat.observers.base_observer import BaseObserver, FramePushed
 
@@ -37,7 +38,7 @@ class MetricsLogObserver(BaseObserver):
     - ProcessingMetricsData (General processing time)
     - LLMUsageMetricsData (Token usage statistics)
     - TTSUsageMetricsData (Text-to-Speech character counts)
-    - SmartTurnMetricsData (Turn prediction metrics)
+    - TurnMetricsData (Turn prediction metrics)
 
     This allows developers to track performance metrics, token usage,
     and other statistics throughout the pipeline.
@@ -70,6 +71,17 @@ class MetricsLogObserver(BaseObserver):
             **kwargs: Additional arguments passed to parent class.
         """
         super().__init__(**kwargs)
+        # Normalize deprecated types in include_metrics
+        if include_metrics and SmartTurnMetricsData in include_metrics:
+            import warnings
+
+            warnings.warn(
+                "SmartTurnMetricsData is deprecated in include_metrics, "
+                "use TurnMetricsData instead.",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+            include_metrics = (include_metrics - {SmartTurnMetricsData}) | {TurnMetricsData}
         self._include_metrics = include_metrics
         self._frames_seen = set()
 
@@ -144,8 +156,8 @@ class MetricsLogObserver(BaseObserver):
             logger.debug(
                 f"📊 {processor_info} TTS USAGE{model_info}: {metrics_data.value} characters at {time_sec:.3f}s"
             )
-        elif isinstance(metrics_data, SmartTurnMetricsData):
-            self._log_smart_turn(metrics_data, processor_info, model_info, time_sec)
+        elif isinstance(metrics_data, TurnMetricsData):
+            self._log_turn(metrics_data, processor_info, model_info, time_sec)
         else:
             # Generic fallback for unknown metrics types
             logger.debug(
@@ -191,28 +203,27 @@ class MetricsLogObserver(BaseObserver):
             f"📊 {processor_info} LLM TOKEN USAGE{model_info}: {usage_str} at {time_sec:.2f}s"
         )
 
-    def _log_smart_turn(
+    def _log_turn(
         self,
-        metrics_data: SmartTurnMetricsData,
+        metrics_data: TurnMetricsData,
         processor_info: str,
         model_info: str,
         time_sec: float,
     ):
-        """Log smart turn prediction metrics.
+        """Log turn prediction metrics.
 
         Args:
-            metrics_data: The smart turn metrics data.
+            metrics_data: The turn metrics data.
             processor_info: Formatted processor name string.
             model_info: Formatted model name string.
             time_sec: Timestamp in seconds.
         """
         complete_str = "COMPLETE" if metrics_data.is_complete else "INCOMPLETE"
+        e2e_str = f"{metrics_data.e2e_processing_time_ms:.1f}ms"
 
         logger.debug(
-            f"📊 {processor_info} SMART TURN{model_info}: {complete_str} "
+            f"📊 {processor_info} TURN{model_info}: {complete_str} "
             f"(probability: {metrics_data.probability:.2%}, "
-            f"inference: {metrics_data.inference_time_ms:.1f}ms, "
-            f"server: {metrics_data.server_total_time_ms:.1f}ms, "
-            f"e2e: {metrics_data.e2e_processing_time_ms:.1f}ms) "
+            f"e2e: {e2e_str}) "
             f"at {time_sec:.2f}s"
         )

src/pipecat/turns/user_stop/turn_analyzer_user_turn_stop_strategy.py

@@ -115,10 +115,14 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy):
         """Handle input audio to check if the turn is completed."""
         state = self._turn_analyzer.append_audio(frame.audio, self._vad_user_speaking)
 
-        # If at this point the model says the turn is complete it will be due to
-        # a timeout, so we mark turn as complete and we trigger the user end of
-        # turn.
+        # Streaming analyzers (e.g. KrispVivaTurn) detect turn completion
+        # frame-by-frame inside append_audio, so COMPLETE is returned here
+        # rather than in analyze_end_of_turn. Batch analyzers (BaseSmartTurn)
+        # return COMPLETE here only on a silence timeout. In either case we
+        # consume and push metrics immediately while they're fresh.
         if state == EndOfTurnState.COMPLETE:
+            _, prediction = await self._turn_analyzer.analyze_end_of_turn()
+            await self._handle_prediction_result(prediction)
             self._turn_complete = True
             await self._maybe_trigger_user_turn_stopped()