diff --git a/changelog/4128.added.md b/changelog/4128.added.md new file mode 100644 index 000000000..7cbeed91b --- /dev/null +++ b/changelog/4128.added.md @@ -0,0 +1 @@ +- Added `on_end_of_turn` event handler to `AssemblyAISTTService`. This fires after the final transcript is pushed, providing a reliable hook for end-of-turn logic that doesn't race with `TranscriptionFrame`. Works in both Pipecat and AssemblyAI turn detection modes. diff --git a/src/pipecat/services/assemblyai/stt.py b/src/pipecat/services/assemblyai/stt.py index fa723a771..8b273188f 100644 --- a/src/pipecat/services/assemblyai/stt.py +++ b/src/pipecat/services/assemblyai/stt.py @@ -129,6 +129,16 @@ class AssemblyAISTTService(WebsocketSTTService): Provides real-time speech transcription using AssemblyAI's WebSocket API. Supports both interim and final transcriptions with configurable parameters for audio processing and connection management. + + Event handlers available (in addition to WebsocketSTTService events): + + - on_end_of_turn(service, transcript): Called when AssemblyAI detects end of turn. + + Example:: + + @service.event_handler("on_end_of_turn") + async def on_end_of_turn(service, transcript): + ... """ Settings = AssemblyAISTTSettings @@ -303,6 +313,8 @@ class AssemblyAISTTService(WebsocketSTTService): self._user_speaking = False + self._register_event_handler("on_end_of_turn") + def _configure_pipecat_turn_mode(self, settings: Settings, is_u3_pro: bool): """Configure settings for Pipecat turn detection mode. @@ -741,6 +753,7 @@ class AssemblyAISTTService(WebsocketSTTService): ) await self._trace_transcription(transcript_text, True, language) await self.stop_processing_metrics() + await self._call_event_handler("on_end_of_turn", transcript_text) else: await self.push_frame( InterimTranscriptionFrame( @@ -774,6 +787,7 @@ class AssemblyAISTTService(WebsocketSTTService): # above, so ordering is preserved) and upstream. await self.broadcast_frame(UserStoppedSpeakingFrame) self._user_speaking = False + await self._call_event_handler("on_end_of_turn", transcript_text) else: await self.push_frame( InterimTranscriptionFrame( diff --git a/src/pipecat/services/deepgram/flux/stt.py b/src/pipecat/services/deepgram/flux/stt.py index d2f44d46f..cb8966769 100644 --- a/src/pipecat/services/deepgram/flux/stt.py +++ b/src/pipecat/services/deepgram/flux/stt.py @@ -53,16 +53,15 @@ class DeepgramFluxSTTService(DeepgramFluxSTTBase, WebsocketService): Event handlers available (in addition to base events): - - on_speech_started(service): Deepgram detected start of speech - - on_utterance_end(service): Deepgram detected end of utterance - - on_end_of_turn(service): Deepgram detected end of turn (EOT) - - on_eager_end_of_turn(service): Deepgram predicted end of turn (EagerEOT) + - on_start_of_turn(service, transcript): Deepgram detected start of speech + - on_end_of_turn(service, transcript): Deepgram detected end of turn (EOT) + - on_eager_end_of_turn(service, transcript): Deepgram predicted end of turn (EagerEOT) - on_turn_resumed(service): User resumed speaking after EagerEOT Example:: @stt.event_handler("on_end_of_turn") - async def on_end_of_turn(service): + async def on_end_of_turn(service, transcript): ... """