Merge pull request #4128 from pipecat-ai/mb/end-of-turn-assembly

Author: markbackman

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.

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 @@ def __init__(
 
         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 @@ async def _handle_transcription(self, message: TurnMessage):
                 )
                 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 @@ async def _handle_transcription(self, message: TurnMessage):
                 # 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(

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):
             ...
     """