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

fix: surface TTSSpeakFrame greetings in on_assistant_turn_stopped

Browse Source 
Two issues were causing TTSSpeakFrame(append_to_context=True) greetings to
silently lose their trailing words and never fire on_assistant_turn_stopped:

- LLMAssistantPushAggregationFrame was emitted without a PTS, so the
  transport routed it through the audio (sync) queue while word-level
  TTSTextFrames travel through the clock queue. The aggregation could reach
  the assistant aggregator before the final words, leaving them orphaned
  in the buffer. Stamp the frame with `_word_last_pts + 1` when there are
  word timestamps so it can't overtake them.

- The aggregator's LLMAssistantPushAggregationFrame handler called
  push_aggregation() directly, bypassing _trigger_assistant_turn_stopped.
  For TTS-only flows there is no LLMFullResponseStartFrame, so the turn
  start timestamp was never set and on_assistant_turn_stopped never fired.
  Open a turn (if needed) and trigger stopped from the handler.

Fixes #4264.
This commit is contained in:
Mark Backman
2026-05-04 10:41:22 -04:00
parent b363b91d12
commit f1a3ee97de
4 changed files with 154 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
src/pipecat/processors/aggregators/llm_response_universal.py
View File

@@ -927,7 +927,7 @@ class LLMAssistantAggregator(LLMContextAggregator):
await self._handle_end_or_cancel(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, LLMAssistantPushAggregationFrame):
await self.push_aggregation()
await self._handle_push_aggregation()
elif isinstance(frame, LLMFullResponseStartFrame):
await self._handle_llm_start(frame)
elif isinstance(frame, LLMFullResponseEndFrame):
@@ -1309,6 +1309,17 @@ class LLMAssistantAggregator(LLMContextAggregator):
async def _handle_llm_end(self, _: LLMFullResponseEndFrame):
await self._trigger_assistant_turn_stopped()
async def _handle_push_aggregation(self):
# LLMAssistantPushAggregationFrame is emitted by TTSService at the end
# of a TTSSpeakFrame-driven utterance (no surrounding LLM response
# cycle), so no LLMFullResponseStartFrame ever set the turn-start
# timestamp. Open a turn now so on_assistant_turn_stopped fires for the
# greeting text the same way it did before LLMAssistantPushAggregationFrame
# was introduced.
if not self._assistant_turn_start_timestamp:
await self._trigger_assistant_turn_started()
await self._trigger_assistant_turn_stopped()
async def _handle_text(self, frame: TextFrame):
# Skip TextFrame types not intended to build the assistant context
if isinstance(frame, (TranscriptionFrame, TranslationFrame, InterimTranscriptionFrame)):

12
src/pipecat/services/tts_service.py
View File

@@ -771,7 +771,17 @@ class TTSService(AIService):
if isinstance(frame, TTSStoppedFrame) and frame.context_id:
if frame.context_id in self._tts_contexts:
if self._tts_contexts[frame.context_id].push_assistant_aggregation:
await self.push_frame(LLMAssistantPushAggregationFrame())
aggregation_frame = LLMAssistantPushAggregationFrame()
# When word-level TTSTextFrames are routed through the
# transport's clock queue (PTS-based), the aggregation frame
# would otherwise take the audio (sync) queue path and
# could overtake the final word frames. Stamping it with a
# PTS just past the last word forces it through the clock
# queue too, so the assistant aggregator sees every word
# before flushing.
if self._word_last_pts:
aggregation_frame.pts = self._word_last_pts + 1
await self.push_frame(aggregation_frame)
logger.debug(f"{self} cleaning up TTS context {frame.context_id}")
del self._tts_contexts[frame.context_id]

80
tests/test_context_aggregators_universal.py
View File

@@ -17,6 +17,7 @@ from pipecat.frames.frames import (
FunctionCallsStartedFrame,
InterimTranscriptionFrame,
InterruptionFrame,
LLMAssistantPushAggregationFrame,
LLMContextAssistantTimestampFrame,
LLMContextFrame,
LLMFullResponseEndFrame,
@@ -34,6 +35,7 @@ from pipecat.frames.frames import (
TextFrame,
TranscriptionFrame,
TranslationFrame,
TTSTextFrame,
UserMuteStartedFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
@@ -58,6 +60,7 @@ from pipecat.turns.user_mute import (
)
from pipecat.turns.user_stop import SpeechTimeoutUserTurnStopStrategy
from pipecat.turns.user_turn_strategies import UserTurnStrategies
from pipecat.utils.text.base_text_aggregator import AggregationType
USER_TURN_STOP_TIMEOUT = 0.2
TRANSCRIPTION_TIMEOUT = 0.1
@@ -975,6 +978,83 @@ class TestLLMAssistantAggregator(unittest.IsolatedAsyncioTestCase):
self.assertEqual(len(stop_messages), 1)
self.assertEqual(stop_messages[0].content, "Hello from Pipecat!")
async def test_push_aggregation_fires_turn_stopped_for_tts_speak(self):
"""LLMAssistantPushAggregationFrame must fire on_assistant_turn_stopped.
Mirrors the TTSSpeakFrame(append_to_context=True) greeting flow: TTS-driven
TTSTextFrames accumulate without an LLMFullResponseStartFrame, then the
TTS service emits LLMAssistantPushAggregationFrame to commit them.
"""
context = LLMContext()
aggregator = LLMAssistantAggregator(context)
start_count = 0
stop_messages = []
@aggregator.event_handler("on_assistant_turn_started")
async def on_assistant_turn_started(aggregator):
nonlocal start_count
start_count += 1
@aggregator.event_handler("on_assistant_turn_stopped")
async def on_assistant_turn_stopped(aggregator, message: AssistantTurnStoppedMessage):
stop_messages.append(message)
frames_to_send = [
TTSTextFrame("Hello,", aggregated_by=AggregationType.WORD),
TTSTextFrame("how", aggregated_by=AggregationType.WORD),
TTSTextFrame("can I help?", aggregated_by=AggregationType.WORD),
LLMAssistantPushAggregationFrame(),
]
expected_down_frames = [LLMContextFrame, LLMContextAssistantTimestampFrame]
await run_test(
aggregator,
frames_to_send=frames_to_send,
expected_down_frames=expected_down_frames,
)
self.assertEqual(start_count, 1)
self.assertEqual(len(stop_messages), 1)
self.assertFalse(stop_messages[0].interrupted)
self.assertEqual(stop_messages[0].content, "Hello, how can I help?")
self.assertEqual(
context.messages[-1],
{"role": "assistant", "content": "Hello, how can I help?"},
)
async def test_push_aggregation_does_not_double_fire_in_llm_response(self):
"""LLMAssistantPushAggregationFrame mid-response must not double-fire turn events.
Inside an LLMFullResponseStart/End cycle, a stray LLMAssistantPushAggregationFrame
should flush whatever is buffered and consume the active turn (firing exactly
one stopped event). The closing LLMFullResponseEndFrame then has no pending
turn to stop.
"""
context = LLMContext()
aggregator = LLMAssistantAggregator(context)
start_count = 0
stop_messages = []
@aggregator.event_handler("on_assistant_turn_started")
async def on_assistant_turn_started(aggregator):
nonlocal start_count
start_count += 1
@aggregator.event_handler("on_assistant_turn_stopped")
async def on_assistant_turn_stopped(aggregator, message: AssistantTurnStoppedMessage):
stop_messages.append(message)
frames_to_send = [
LLMFullResponseStartFrame(),
LLMTextFrame("Hello!"),
LLMAssistantPushAggregationFrame(),
LLMFullResponseEndFrame(),
]
await run_test(aggregator, frames_to_send=frames_to_send)
self.assertEqual(start_count, 1)
self.assertEqual(len(stop_messages), 1)
self.assertEqual(stop_messages[0].content, "Hello!")
async def test_turn_completion_markers_stripped_from_transcript(self):
"""Turn completion markers should be stripped from assistant transcript."""
from pipecat.turns.user_turn_completion_mixin import (

51
tests/test_tts_frame_ordering.py
View File

@@ -33,6 +33,7 @@ from pipecat.frames.frames import (
AggregatedTextFrame,
DataFrame,
Frame,
LLMAssistantPushAggregationFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
TextFrame,
@@ -657,5 +658,55 @@ async def test_websocket_word_timestamps_punctuation_tokens():
assert all(f.includes_inter_frame_spaces is True for f in text_frames)
@pytest.mark.asyncio
async def test_push_aggregation_pts_after_last_word():
"""LLMAssistantPushAggregationFrame must carry PTS > last TTSTextFrame PTS.
Without a PTS the aggregation frame routes through the transport's audio
(sync) queue while word-level TTSTextFrames go through the clock queue, so
it can overtake the last words and leave the trailing text orphaned in the
aggregator buffer (issue #4264).
"""
word_times = [("hello", 0.0), ("world", 0.2)]
tts = _MockWordTimestampHttpTTSService(word_times=word_times)
frames_received = await run_test(
tts,
frames_to_send=[TTSSpeakFrame(text="hello world", append_to_context=True)],
)
down = frames_received[0]
text_frames = [f for f in down if isinstance(f, TTSTextFrame)]
push_frames = [f for f in down if isinstance(f, LLMAssistantPushAggregationFrame)]
assert len(push_frames) == 1, (
f"Expected exactly one LLMAssistantPushAggregationFrame, got {len(push_frames)}"
)
assert text_frames, "Expected TTSTextFrames to be emitted"
last_word_pts = max(f.pts for f in text_frames)
assert push_frames[0].pts is not None and push_frames[0].pts > last_word_pts, (
f"LLMAssistantPushAggregationFrame.pts ({push_frames[0].pts}) must exceed "
f"the last TTSTextFrame PTS ({last_word_pts}) so it can't overtake it in the "
f"transport's clock queue"
)
@pytest.mark.asyncio
async def test_push_aggregation_no_pts_without_word_timestamps():
"""Aggregation frame stays unstamped when no word timestamps were emitted.
Without word frames, both the aggregation frame and any other downstream
frames travel through the transport's sync queue in order, so adding a PTS
would needlessly route it through the clock queue.
"""
tts = MockHttpTTSService()
frames_received = await run_test(
tts,
frames_to_send=[TTSSpeakFrame(text="hello world", append_to_context=True)],
)
push_frames = [f for f in frames_received[0] if isinstance(f, LLMAssistantPushAggregationFrame)]
assert len(push_frames) == 1
assert push_frames[0].pts is None or push_frames[0].pts == 0
if __name__ == "__main__":
unittest.main()