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

Add Krisp API key support and debug logging

Author: markbackman

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.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.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 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 @@ def EVAL_VISION_IMAGE(*, eval_speaks_first: bool = False):
     ("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 @@ def EVAL_VISION_IMAGE(*, eval_speaks_first: bool = False):
     ("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 @@ def __init__(
                 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 @@ def __init__(
         """
         super().__init__()
 
+        self._api_key = api_key
+
         try:
             # Set model path, checking environment if not specified
             if model_path:
@@ -132,7 +140,7 @@ async def start(self, sample_rate: int):
         """
         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,25 +89,46 @@ 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)
         """
         with cls._lock:
             # 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 @@
 """
 
 import os
+import time
 from typing import Optional, Tuple
 
 import numpy as np
@@ -26,7 +27,7 @@
     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 @@ def __init__(
         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 @@ def __init__(
             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 @@ def __init__(
 
         # 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 @@ def __init__(
             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 @@ def append_audio(self, buffer: bytes, is_speech: bool) -> EndOfTurnState:
                     # 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 @@ def append_audio(self, buffer: bytes, is_speech: bool) -> EndOfTurnState:
                 # 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 @@ async def analyze_end_of_turn(self) -> Tuple[EndOfTurnState, Optional[MetricsDat
             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