wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add TurnMetricsData and e2e processing time for KrispVivaTurn

Browse Source 
Introduce a generic TurnMetricsData class for turn detection metrics,
replacing the service-specific SmartTurnMetricsData (now deprecated).
Add end-to-end processing time measurement to KrispVivaTurn, tracking
the interval from VAD speech-to-silence transition to model threshold
crossing. Consume metrics in the strategy _handle_input_audio path
so they are pushed immediately when fresh.
This commit is contained in:
Mark Backman
2026-02-24 17:42:18 -05:00
parent 73ee4da7d4
commit 0ca8c850fb
10 changed files with 92 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
changelog/3809.added.md
View File

@@ -1 +1 @@
- Added `api_key` parameter to `KrispVivaSDKManager`, `KrispVivaTurn`, and `KrispVivaFilter` for Krisp SDK v1.8.0 licensing. Falls back to `KRISP_API_KEY` environment variable. Backwards compatible with older SDK versions.
- 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.

2
changelog/3809.changed.md
View File

@@ -1 +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.
- 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.

1
changelog/3809.deprecated.md Normal file
View File

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

3
examples/foundational/07p-interruptible-krisp-viva.py
View File

@@ -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
@@ -124,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")

13
examples/foundational/38b-smart-turn-local.py
View File

@@ -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):

37
src/pipecat/audio/turn/krisp_viva_turn.py
View File

@@ -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
@@ -118,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
@@ -291,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.
@@ -311,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
@@ -332,15 +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
logger.debug(
f"Krisp turn analysis: state={self._last_state}, probability={self._last_probability}"
)
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

13
src/pipecat/audio/turn/smart_turn/base_smart_turn.py
View File

@@ -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(

26
src/pipecat/metrics/metrics.py
View File

@@ -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

33
src/pipecat/observers/loggers/metrics_log_observer.py
View File

@@ -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"
)

10
src/pipecat/turns/user_stop/turn_analyzer_user_turn_stop_strategy.py
View File

@@ -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()