diff --git a/examples/realtime/realtime-gemini-live-local-vad.py b/examples/realtime/realtime-gemini-live-local-vad.py index 7a73742e9..288b24eee 100644 --- a/examples/realtime/realtime-gemini-live-local-vad.py +++ b/examples/realtime/realtime-gemini-live-local-vad.py @@ -10,6 +10,7 @@ import os from dotenv import load_dotenv from loguru import logger +from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3 from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.frames.frames import LLMRunFrame from pipecat.pipeline.pipeline import Pipeline @@ -28,6 +29,9 @@ from pipecat.services.google.gemini_live.llm import GeminiLiveLLMService, Gemini 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_start import VADUserTurnStartStrategy +from pipecat.turns.user_stop import VADTurnAnalyzerUserTurnStopStrategy +from pipecat.turns.user_turn_strategies import UserTurnStrategies load_dotenv(override=True) @@ -73,6 +77,12 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): user_aggregator, assistant_aggregator = LLMContextAggregatorPair( context, user_params=LLMUserAggregatorParams( + user_turn_strategies=UserTurnStrategies( + start=[VADUserTurnStartStrategy()], + stop=[ + VADTurnAnalyzerUserTurnStopStrategy(turn_analyzer=LocalSmartTurnAnalyzerV3()) + ], + ), vad_analyzer=SileroVADAnalyzer(), ), ) diff --git a/src/pipecat/turns/user_stop/__init__.py b/src/pipecat/turns/user_stop/__init__.py index 7ff676744..6bac6ccc4 100644 --- a/src/pipecat/turns/user_stop/__init__.py +++ b/src/pipecat/turns/user_stop/__init__.py @@ -8,11 +8,13 @@ from .base_user_turn_stop_strategy import BaseUserTurnStopStrategy, UserTurnStop from .external_user_turn_stop_strategy import ExternalUserTurnStopStrategy from .speech_timeout_user_turn_stop_strategy import SpeechTimeoutUserTurnStopStrategy from .turn_analyzer_user_turn_stop_strategy import TurnAnalyzerUserTurnStopStrategy +from .vad_turn_analyzer_user_turn_stop_strategy import VADTurnAnalyzerUserTurnStopStrategy __all__ = [ "BaseUserTurnStopStrategy", "ExternalUserTurnStopStrategy", "SpeechTimeoutUserTurnStopStrategy", - "UserTurnStoppedParams", "TurnAnalyzerUserTurnStopStrategy", + "UserTurnStoppedParams", + "VADTurnAnalyzerUserTurnStopStrategy", ] diff --git a/src/pipecat/turns/user_stop/vad_turn_analyzer_user_turn_stop_strategy.py b/src/pipecat/turns/user_stop/vad_turn_analyzer_user_turn_stop_strategy.py new file mode 100644 index 000000000..03f868aae --- /dev/null +++ b/src/pipecat/turns/user_stop/vad_turn_analyzer_user_turn_stop_strategy.py @@ -0,0 +1,132 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""User turn stop strategy based on VAD and turn detection analyzers. + +This strategy uses a turn analyzer to detect end-of-turn but does not +require STT transcriptions. It triggers immediately when the turn analyzer +indicates COMPLETE, making it suitable for speech-to-speech pipelines +where transcriptions arrive too late to be useful for turn decisions. +""" + +from typing import Optional + +from pipecat.audio.turn.base_turn_analyzer import BaseTurnAnalyzer, EndOfTurnState +from pipecat.frames.frames import ( + Frame, + InputAudioRawFrame, + MetricsFrame, + SpeechControlParamsFrame, + StartFrame, + VADUserStartedSpeakingFrame, + VADUserStoppedSpeakingFrame, +) +from pipecat.metrics.metrics import MetricsData +from pipecat.turns.types import ProcessFrameResult +from pipecat.turns.user_stop.base_user_turn_stop_strategy import BaseUserTurnStopStrategy +from pipecat.utils.asyncio.task_manager import BaseTaskManager + + +class VADTurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy): + """User turn stop strategy that uses a turn analyzer without waiting for transcriptions. + + This strategy feeds audio and VAD frames to a turn detection model + (``BaseTurnAnalyzer``) and triggers immediately when the model indicates + the turn is complete. Unlike ``TurnAnalyzerUserTurnStopStrategy``, it does + not wait for STT transcriptions, making it ideal for speech-to-speech + pipelines (e.g. Gemini Live) where audio goes directly to the LLM. + + The ``UserTurnController`` provides a safety-net timeout + (``user_turn_stop_timeout``, default 5s) if the turn analyzer never + returns COMPLETE. + """ + + def __init__(self, *, turn_analyzer: BaseTurnAnalyzer, **kwargs): + """Initialize the user turn stop strategy. + + Args: + turn_analyzer: The turn detection analyzer instance to detect end of user turn. + **kwargs: Additional keyword arguments. + """ + super().__init__(**kwargs) + self._turn_analyzer = turn_analyzer + self._vad_user_speaking = False + + async def reset(self): + """Reset the strategy to its initial state.""" + await super().reset() + self._vad_user_speaking = False + + async def setup(self, task_manager: BaseTaskManager): + """Initialize the strategy with the given task manager. + + Args: + task_manager: The task manager to be associated with this instance. + """ + await super().setup(task_manager) + + async def cleanup(self): + """Cleanup the strategy.""" + await super().cleanup() + await self._turn_analyzer.cleanup() + + async def process_frame(self, frame: Frame) -> ProcessFrameResult: + """Process an incoming frame to update the turn analyzer and strategy state. + + Args: + frame: The frame to be analyzed. + + Returns: + Always returns CONTINUE so subsequent stop strategies are evaluated. + """ + await super().process_frame(frame) + + if isinstance(frame, StartFrame): + await self._start(frame) + elif isinstance(frame, VADUserStartedSpeakingFrame): + await self._handle_vad_user_started_speaking(frame) + elif isinstance(frame, VADUserStoppedSpeakingFrame): + await self._handle_vad_user_stopped_speaking(frame) + elif isinstance(frame, InputAudioRawFrame): + await self._handle_input_audio(frame) + + return ProcessFrameResult.CONTINUE + + async def _start(self, frame: StartFrame): + """Process the start frame to configure the turn analyzer.""" + self._turn_analyzer.set_sample_rate(frame.audio_in_sample_rate) + await self.broadcast_frame(SpeechControlParamsFrame, turn_params=self._turn_analyzer.params) + + async def _handle_input_audio(self, frame: InputAudioRawFrame): + """Handle input audio to check if the turn is completed.""" + state = self._turn_analyzer.append_audio(frame.audio, self._vad_user_speaking) + + # Streaming analyzers (e.g. KrispVivaTurn) detect turn completion + # frame-by-frame inside append_audio, so COMPLETE is returned here. + if state == EndOfTurnState.COMPLETE: + _, prediction = await self._turn_analyzer.analyze_end_of_turn() + await self._handle_prediction_result(prediction) + await self.trigger_user_turn_stopped() + + async def _handle_vad_user_started_speaking(self, frame: VADUserStartedSpeakingFrame): + """Handle when the VAD indicates the user is speaking.""" + self._turn_analyzer.update_vad_start_secs(frame.start_secs) + self._vad_user_speaking = True + + async def _handle_vad_user_stopped_speaking(self, frame: VADUserStoppedSpeakingFrame): + """Handle when the VAD indicates the user has stopped speaking.""" + self._vad_user_speaking = False + + state, prediction = await self._turn_analyzer.analyze_end_of_turn() + await self._handle_prediction_result(prediction) + + if state == EndOfTurnState.COMPLETE: + await self.trigger_user_turn_stopped() + + async def _handle_prediction_result(self, result: Optional[MetricsData]): + """Handle a prediction result event from the turn analyzer.""" + if result: + await self.push_frame(MetricsFrame(data=[result])) diff --git a/tests/test_vad_turn_analyzer_user_turn_stop_strategy.py b/tests/test_vad_turn_analyzer_user_turn_stop_strategy.py new file mode 100644 index 000000000..add762b0d --- /dev/null +++ b/tests/test_vad_turn_analyzer_user_turn_stop_strategy.py @@ -0,0 +1,214 @@ +# +# Copyright (c) 2024-2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Tests for VADTurnAnalyzerUserTurnStopStrategy.""" + +import unittest +from typing import Optional, Tuple +from unittest.mock import AsyncMock + +import pytest + +from pipecat.audio.turn.base_turn_analyzer import ( + BaseTurnAnalyzer, + BaseTurnParams, + EndOfTurnState, +) +from pipecat.frames.frames import ( + InputAudioRawFrame, + MetricsFrame, + SpeechControlParamsFrame, + StartFrame, + TranscriptionFrame, + VADUserStartedSpeakingFrame, + VADUserStoppedSpeakingFrame, +) +from pipecat.metrics.metrics import MetricsData +from pipecat.turns.user_stop.vad_turn_analyzer_user_turn_stop_strategy import ( + VADTurnAnalyzerUserTurnStopStrategy, +) +from pipecat.utils.asyncio.task_manager import TaskManager + + +class MockTurnAnalyzer(BaseTurnAnalyzer): + """Mock turn analyzer for testing.""" + + def __init__(self): + super().__init__() + self._append_audio_state = EndOfTurnState.INCOMPLETE + self._analyze_state = EndOfTurnState.INCOMPLETE + self._prediction: Optional[MetricsData] = None + self._speech_triggered = False + + @property + def speech_triggered(self) -> bool: + return self._speech_triggered + + @property + def params(self) -> BaseTurnParams: + return BaseTurnParams() + + def append_audio(self, buffer: bytes, is_speech: bool) -> EndOfTurnState: + return self._append_audio_state + + async def analyze_end_of_turn(self) -> Tuple[EndOfTurnState, Optional[MetricsData]]: + return self._analyze_state, self._prediction + + def update_vad_start_secs(self, vad_start_secs: float): + pass + + def clear(self): + pass + + +class TestVADTurnAnalyzerUserTurnStopStrategy(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + self.analyzer = MockTurnAnalyzer() + self.strategy = VADTurnAnalyzerUserTurnStopStrategy(turn_analyzer=self.analyzer) + self.task_manager = TaskManager() + await self.strategy.setup(self.task_manager) + + self.turn_stopped_called = False + + @self.strategy.event_handler("on_user_turn_stopped") + async def on_user_turn_stopped(strategy, params): + self.turn_stopped_called = True + + self.pushed_frames = [] + + @self.strategy.event_handler("on_push_frame") + async def on_push_frame(strategy, frame, direction): + self.pushed_frames.append(frame) + + self.broadcast_frames = [] + + @self.strategy.event_handler("on_broadcast_frame") + async def on_broadcast_frame(strategy, frame_cls, **kwargs): + self.broadcast_frames.append((frame_cls, kwargs)) + + async def asyncTearDown(self): + await self.strategy.cleanup() + + async def test_vad_stop_complete_triggers_immediately(self): + """VAD stop with COMPLETE should trigger user turn stopped immediately.""" + self.analyzer._analyze_state = EndOfTurnState.COMPLETE + + await self.strategy.process_frame(VADUserStartedSpeakingFrame(start_secs=0.2)) + await self.strategy.process_frame(VADUserStoppedSpeakingFrame(stop_secs=0.3)) + + assert self.turn_stopped_called + + async def test_vad_stop_incomplete_does_not_trigger(self): + """VAD stop with INCOMPLETE should not trigger user turn stopped.""" + self.analyzer._analyze_state = EndOfTurnState.INCOMPLETE + + await self.strategy.process_frame(VADUserStartedSpeakingFrame(start_secs=0.2)) + await self.strategy.process_frame(VADUserStoppedSpeakingFrame(stop_secs=0.3)) + + assert not self.turn_stopped_called + + async def test_streaming_complete_via_append_audio_triggers(self): + """Streaming COMPLETE from append_audio should trigger immediately.""" + self.analyzer._append_audio_state = EndOfTurnState.COMPLETE + # analyze_end_of_turn is called after append_audio returns COMPLETE + self.analyzer._analyze_state = EndOfTurnState.COMPLETE + + await self.strategy.process_frame(VADUserStartedSpeakingFrame(start_secs=0.2)) + await self.strategy.process_frame( + InputAudioRawFrame(audio=b"\x00" * 320, sample_rate=16000, num_channels=1) + ) + + assert self.turn_stopped_called + + async def test_streaming_incomplete_does_not_trigger(self): + """Streaming INCOMPLETE from append_audio should not trigger.""" + self.analyzer._append_audio_state = EndOfTurnState.INCOMPLETE + + await self.strategy.process_frame( + InputAudioRawFrame(audio=b"\x00" * 320, sample_rate=16000, num_channels=1) + ) + + assert not self.turn_stopped_called + + async def test_vad_start_resets_speaking_state(self): + """VAD start should set speaking state to True.""" + assert not self.strategy._vad_user_speaking + + await self.strategy.process_frame(VADUserStartedSpeakingFrame(start_secs=0.2)) + + assert self.strategy._vad_user_speaking + + async def test_vad_stop_resets_speaking_state(self): + """VAD stop should set speaking state to False.""" + self.analyzer._analyze_state = EndOfTurnState.INCOMPLETE + + await self.strategy.process_frame(VADUserStartedSpeakingFrame(start_secs=0.2)) + assert self.strategy._vad_user_speaking + + await self.strategy.process_frame(VADUserStoppedSpeakingFrame(stop_secs=0.3)) + assert not self.strategy._vad_user_speaking + + async def test_reset_clears_state(self): + """Reset should clear speaking state.""" + self.strategy._vad_user_speaking = True + + await self.strategy.reset() + + assert not self.strategy._vad_user_speaking + + async def test_metrics_frame_pushed_on_prediction(self): + """MetricsFrame should be pushed when prediction result is available.""" + prediction = MetricsData(processor="test_analyzer") + self.analyzer._analyze_state = EndOfTurnState.COMPLETE + self.analyzer._prediction = prediction + + await self.strategy.process_frame(VADUserStartedSpeakingFrame(start_secs=0.2)) + await self.strategy.process_frame(VADUserStoppedSpeakingFrame(stop_secs=0.3)) + + assert len(self.pushed_frames) == 1 + assert isinstance(self.pushed_frames[0], MetricsFrame) + assert self.pushed_frames[0].data == [prediction] + + async def test_no_metrics_frame_when_no_prediction(self): + """No MetricsFrame should be pushed when prediction is None.""" + self.analyzer._analyze_state = EndOfTurnState.COMPLETE + self.analyzer._prediction = None + + await self.strategy.process_frame(VADUserStartedSpeakingFrame(start_secs=0.2)) + await self.strategy.process_frame(VADUserStoppedSpeakingFrame(stop_secs=0.3)) + + assert len(self.pushed_frames) == 0 + + async def test_speech_control_params_broadcast_on_start(self): + """SpeechControlParamsFrame should be broadcast on StartFrame.""" + await self.strategy.process_frame( + StartFrame(audio_in_sample_rate=16000, audio_out_sample_rate=16000) + ) + + assert len(self.broadcast_frames) == 1 + frame_cls, kwargs = self.broadcast_frames[0] + assert frame_cls is SpeechControlParamsFrame + + async def test_transcription_frames_ignored(self): + """TranscriptionFrame should not affect state or trigger turn stop.""" + self.analyzer._analyze_state = EndOfTurnState.INCOMPLETE + + await self.strategy.process_frame( + TranscriptionFrame(text="hello", user_id="user1", timestamp="now") + ) + + assert not self.turn_stopped_called + + async def test_process_frame_returns_continue(self): + """process_frame should always return CONTINUE.""" + from pipecat.turns.types import ProcessFrameResult + + result = await self.strategy.process_frame(VADUserStartedSpeakingFrame(start_secs=0.2)) + assert result == ProcessFrameResult.CONTINUE + + +if __name__ == "__main__": + unittest.main()