TranscriptionFrame And UserStoppedSpeakingFrame Ordering Not Preserved

[jcbjoe]

pipecat version

0.0.105

Python version

3.11.5

Operating System

MacOS 26.2 (25C56)

Issue description

For Deepgram Flux and Assembly AI it seems the ordering of TranscriptionFrame and UserStoppedSpeakingFrame is not Preserved. There is even a comment in the AssemblyAI processor saying it should be:

                # AAI is authoritative — emit UserStoppedSpeakingFrame immediately.
                # broadcast_frame pushes downstream (same queue as TranscriptionFrame
                # above, so ordering is preserved) and upstream.

However on testing I find that the UserStoppedSpeakingFrame reaches my subsequent processors first then the TranscriptionFrame gets received later.

I have done some testing on this and I believe its down to UserStoppedSpeakingFrame being a SystemFrame. I tried this with another frame that I made "EndOfTurnDetectedFrame"

@dataclass
class EndOfTurnDetectedFrame(DataFrame):
    pass

Which seems to work fine.

Here is a snippet from the DeepgramFluxProcessor and you can see that the TranscriptionFrame is first then the UserStoppedSpeakingFrame is broadcast after

        if not self._settings.min_confidence or average_confidence > self._settings.min_confidence:
            # EndOfTurn means Flux has determined the turn is complete,
            # so this TranscriptionFrame is always finalized
            await self.push_frame(
                TranscriptionFrame(
                    transcript,
                    self._user_id,
                    time_now_iso8601(),
                    self._settings.language,
                    result=data,
                    finalized=True,
                )
            )
        else:
            logger.warning(
                f"Transcription confidence below min_confidence threshold: {average_confidence}"
            )

        await self._handle_transcription(transcript, True, self._settings.language)
        await self.stop_processing_metrics()
        await self.broadcast_frame(UserStoppedSpeakingFrame)
        await self._call_event_handler("on_end_of_turn", transcript)

Reproduction steps

1. Make a test processor, in my case I called it OrderingProcessor

from loguru import logger
from pipecat.frames.frames import UserStoppedSpeakingFrame, Frame, TranscriptionFrame
from pipecat.processors.frame_processor import FrameProcessor, FrameDirection

class OrderingProcessor(FrameProcessor):
    async def process_frame(self, frame: Frame, direction: FrameDirection):
        await super().process_frame(frame, direction)
        if isinstance(frame, TranscriptionFrame):
            logger.info("Received TranscriptionFrame")
        if isinstance(frame, UserStoppedSpeakingFrame):
            logger.info("Received UserStoppedSpeakingFrame")
        await self.push_frame(frame, direction)

2. Then make a simple pipeline with both Deepgram and Ordering Processor
   For example:

        pipeline = Pipeline(
            [
                transport.input(), 
                DeepgramFluxSTTService(....),
                OrderingProcessor(),
                transport.output(), 
            ]
        )

3. Check the logs and you will see that UserStoppedSpeakingFrame reaches first

{"timestamp": 1773314786.697549, "level": "INFO", "message": "Received UserStoppedSpeakingFrame"}
{"timestamp": 1773314786.698449, "level": "INFO", "message": "Received TranscriptionFrame"}

Expected behavior

Expected behaviour is that TranscriptionFrame will reach the following processors first

Actual behavior

Actual behaviour is that UserStoppedSpeakingFrame will reach the following processors first

Workaround

Deepgram has a on_end_of_turn event I can use to send my own EndOfTurnDetectedFrame that isnt a system frame which works. However AssemblyAI does not have the same event sadly

await self._call_event_handler("on_end_of_turn", transcript)

[markbackman]

Your analysis seems correct. Yes, the UserStoppedSpeakingFrame is a SystemFrame, so it can eventually get ahead of the TranscriptionFrame, thought the TranscriptionFrame is consumed by the LLMUserAggregator, which is usually immediately downstream of the STTService.

Would you mind explaining what you're doing (or trying to do) and why this is impacting you?

[jcbjoe]

Hey Mark for our use case we don't actually use the LLMUserAggregator. We have a custom frame processor which decides what to say/do/execute. Sometimes going to an LLM but sometimes not. We were trying to use the UserStoppedSpeakingFrame as a way to signal to our platform an end of turn event had been triggered.

[markbackman]

Thanks, that helps. In the LLMUserAggregator, there are some user stop strategies that look for multiple signals to be received to declare an end of turn. In all cases, the transcription text must be present. I'd recommend following a similar pattern. For example, you can see this logic at play in the ExternalUserTurnStopStrategy, which you'd use if you have an LLMUserAggregator.
https://github.com/pipecat-ai/pipecat/blob/main/src/pipecat/turns/user_stop/external_user_turn_stop_strategy.py

You can see that, once the UserStoppedSpeakingFrame is received, it calls _maybe_trigger_user_turn_stopped(), which only stops the turn if transcription text has been received (among other things). This is a safety mechanism to make sure that there is text available to trigger an LLM inference.

As for the frame ordering, this is something we'll have to think about. Changing UserStoppedSpeakingFrame to something other than SystemFrame will have consequences, so that feels like not the right call. It's possible that the ordering of these two frames isn't or can't be guaranteed. (The comment in AssemblyAI is from a non-maintainer, so we should fix it.)