diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 505c9ca1d..569574842 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -339,6 +339,40 @@ class LLMTextFrame(TextFrame): self.includes_inter_frame_spaces = True +@dataclass +class LLMMarkerFrame(DataFrame): + """Sideband marker emitted by an LLM service. + + A marker is short, structured assistant output that should be + persisted in the conversation context but should not flow through + the standard text path (TTS, transcript). The assistant aggregator + writes the marker to the context so the LLM can self-condition on + prior markers on subsequent turns. + + The primary use today is the ``filter_incomplete_user_turns`` + protocol, where ``UserTurnCompletionLLMServiceMixin`` emits the + turn-completion markers ✓ / ○ / ◐ on every response. The frame is + intentionally generic so other components — STT services with + built-in turn signals, end-of-turn classifiers, custom annotations, + etc. — can use the same mechanism to inject sideband signals into + the assistant context. + + Parameters: + marker: The marker payload (typically a short string such as a + single character). + append_to_context_immediately: If True, the marker is written + to the context as its own standalone assistant message as + soon as it's received. If False, the marker is appended to + the running assistant aggregation and flushed to the + context together with the following text as a single + message (e.g. for the ✓ case the context message ends up + as "✓ "). + """ + + marker: str + append_to_context_immediately: bool = True + + @dataclass class AggregatedTextFrame(TextFrame): """Text frame representing an aggregation of TextFrames. diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index 1f61f33bb..019490929 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -44,6 +44,7 @@ from pipecat.frames.frames import ( LLMContextSummaryRequestFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, + LLMMarkerFrame, LLMMessagesAppendFrame, LLMMessagesTransformFrame, LLMMessagesUpdateFrame, @@ -1105,6 +1106,8 @@ class LLMAssistantAggregator(LLMContextAggregator): await self._handle_llm_end(frame) elif isinstance(frame, TextFrame): await self._handle_text(frame) + elif isinstance(frame, LLMMarkerFrame): + await self._handle_marker_frame(frame) elif isinstance(frame, LLMThoughtStartFrame): await self._handle_thought_start(frame) elif isinstance(frame, LLMThoughtTextFrame): @@ -1510,6 +1513,31 @@ class LLMAssistantAggregator(LLMContextAggregator): ) ) + async def _handle_marker_frame(self, frame: LLMMarkerFrame): + if frame.append_to_context_immediately: + # Stand-alone marker: write it to the context now as its + # own assistant message. Used when the marker is the entire + # assistant turn — e.g. the ○ / ◐ incomplete-turn signals, + # where the spoken response is suppressed and the marker + # is the only artifact. + self._context.add_message({"role": "assistant", "content": frame.marker}) + await self.push_context_frame() + timestamp_frame = LLMContextAssistantTimestampFrame(timestamp=time_now_iso8601()) + await self.push_frame(timestamp_frame) + return + + # Marker is part of an in-progress assistant response. Append + # it to the running aggregation so `push_aggregation` writes + # marker + text as a single context message — e.g. the ✓ + # complete-turn signal that prefixes the spoken response, + # producing "✓ " in context. Markers are stripped + # from the transcript via + # `_maybe_strip_turn_completion_markers` so consumers see + # clean text. + self._aggregation.append( + TextPartForConcatenation(frame.marker, includes_inter_part_spaces=False) + ) + async def _handle_thought_start(self, frame: LLMThoughtStartFrame): await self._reset_thought_aggregation() self._thought_append_to_context = frame.append_to_context diff --git a/src/pipecat/turns/user_turn_completion_mixin.py b/src/pipecat/turns/user_turn_completion_mixin.py index 606ba9902..50ead0b82 100644 --- a/src/pipecat/turns/user_turn_completion_mixin.py +++ b/src/pipecat/turns/user_turn_completion_mixin.py @@ -22,6 +22,7 @@ from pipecat.frames.frames import ( Frame, InterruptionFrame, LLMFullResponseEndFrame, + LLMMarkerFrame, LLMMessagesAppendFrame, LLMRunFrame, LLMTextFrame, @@ -407,11 +408,11 @@ class UserTurnCompletionLLMServiceMixin(FrameProcessor): # explicitly not complete. The re-prompt path is driven by # this mixin's own timeout. - # Push the marker with skip_tts=True so it's added to context (maintains - # conversation continuity per prompt instructions) but not spoken by TTS - frame = LLMTextFrame(self._turn_text_buffer) - frame.skip_tts = True - await self.push_frame(frame) + # Persist the marker to context as a stand-alone assistant + # message via LLMMarkerFrame: the bot produces no spoken + # output for incomplete turns, so the marker is the entire + # context entry. + await self.push_frame(LLMMarkerFrame(marker)) self._turn_text_buffer = "" await self._start_incomplete_timeout(incomplete_type) @@ -424,21 +425,22 @@ class UserTurnCompletionLLMServiceMixin(FrameProcessor): # Broadcast that the user turn is complete so a stop strategy # gating finalization on this signal (e.g. # LLMTurnCompletionUserTurnStopStrategy) can fire - # `on_user_turn_stopped`. Must fire before the LLMTextFrame so + # `on_user_turn_stopped`. Must fire before the marker so # downstream consumers see the signal before the response. await self.broadcast_frame(UserTurnCompletedFrame) + # Push the marker as a sideband signal that the assistant + # aggregator will prepend to the upcoming aggregated text, + # so the context message ends up as "✓ ". + await self.push_frame( + LLMMarkerFrame(USER_TURN_COMPLETE_MARKER, append_to_context_immediately=False) + ) + # Split buffer at the marker to handle cases where marker and text # arrive in the same chunk (e.g., "✓ Hello!" from some LLMs) marker_pos = self._turn_text_buffer.index(USER_TURN_COMPLETE_MARKER) marker_end = marker_pos + len(USER_TURN_COMPLETE_MARKER) - # Push the marker with skip_tts=True - adds to context but not spoken - marker_text = self._turn_text_buffer[:marker_end] - frame = LLMTextFrame(marker_text) - frame.skip_tts = True - await self.push_frame(frame) - # Push remaining text after marker as normal speech remaining_text = self._turn_text_buffer[marker_end:] if remaining_text: diff --git a/tests/test_user_turn_completion_mixin.py b/tests/test_user_turn_completion_mixin.py index 66a48b51f..0947d0a3c 100644 --- a/tests/test_user_turn_completion_mixin.py +++ b/tests/test_user_turn_completion_mixin.py @@ -8,7 +8,12 @@ import unittest import unittest.mock from unittest.mock import AsyncMock -from pipecat.frames.frames import LLMFullResponseEndFrame, LLMTextFrame, UserTurnCompletedFrame +from pipecat.frames.frames import ( + LLMFullResponseEndFrame, + LLMMarkerFrame, + LLMTextFrame, + UserTurnCompletedFrame, +) from pipecat.processors.frame_processor import FrameProcessor from pipecat.services.llm_service import LLMService from pipecat.services.settings import LLMSettings @@ -44,25 +49,24 @@ class TestUserUserTurnCompletionLLMServiceMixin(unittest.IsolatedAsyncioTestCase # Simulate LLM generating: "✓ Hello there!" await processor._push_turn_text(f"{USER_TURN_COMPLETE_MARKER} Hello there!") - # Two LLMTextFrames: marker (skip_tts) and content (normal). The - # broadcast also pushes UserTurnCompletedFrame upstream + downstream. + # The marker rides as LLMMarkerFrame(append_to_context_immediately=False); + # only the spoken text is pushed as an LLMTextFrame. text_frames = [f for f in pushed_frames if isinstance(f, LLMTextFrame)] - self.assertEqual(len(text_frames), 2) + self.assertEqual(len(text_frames), 1) + self.assertEqual(text_frames[0].text, "Hello there!") + self.assertFalse(text_frames[0].skip_tts) - # First text frame should be the marker with skip_tts=True - self.assertEqual(text_frames[0].text, USER_TURN_COMPLETE_MARKER) - self.assertTrue(text_frames[0].skip_tts) - - # Second text frame should be the actual text without skip_tts - self.assertEqual(text_frames[1].text, "Hello there!") - self.assertFalse(text_frames[1].skip_tts) + marker_frames = [f for f in pushed_frames if isinstance(f, LLMMarkerFrame)] + self.assertEqual(len(marker_frames), 1) + self.assertEqual(marker_frames[0].marker, USER_TURN_COMPLETE_MARKER) + self.assertFalse(marker_frames[0].append_to_context_immediately) # UserTurnCompletedFrame broadcast in both directions. completed = [f for f in pushed_frames if isinstance(f, UserTurnCompletedFrame)] self.assertEqual(len(completed), 2) async def test_incomplete_short_marker_suppresses_text(self): - """Test that ○ marker suppresses text with skip_tts and emits no completed frame.""" + """Test that ○ marker suppresses text and is emitted as a stand-alone marker frame.""" processor = MockProcessor() pushed_frames = [] @@ -74,17 +78,21 @@ class TestUserUserTurnCompletionLLMServiceMixin(unittest.IsolatedAsyncioTestCase await processor._push_turn_text(USER_TURN_INCOMPLETE_SHORT_MARKER) + # No LLMTextFrame: response is suppressed. text_frames = [f for f in pushed_frames if isinstance(f, LLMTextFrame)] - self.assertEqual(len(text_frames), 1) - self.assertEqual(text_frames[0].text, USER_TURN_INCOMPLETE_SHORT_MARKER) - self.assertTrue(text_frames[0].skip_tts) + self.assertEqual(len(text_frames), 0) + + marker_frames = [f for f in pushed_frames if isinstance(f, LLMMarkerFrame)] + self.assertEqual(len(marker_frames), 1) + self.assertEqual(marker_frames[0].marker, USER_TURN_INCOMPLETE_SHORT_MARKER) + self.assertTrue(marker_frames[0].append_to_context_immediately) # Incomplete markers do not emit UserTurnCompletedFrame. completed = [f for f in pushed_frames if isinstance(f, UserTurnCompletedFrame)] self.assertEqual(len(completed), 0) async def test_incomplete_long_marker_suppresses_text(self): - """Test that ◐ marker suppresses text with skip_tts and emits no completed frame.""" + """Test that ◐ marker suppresses text and is emitted as a stand-alone marker frame.""" processor = MockProcessor() pushed_frames = [] @@ -97,9 +105,12 @@ class TestUserUserTurnCompletionLLMServiceMixin(unittest.IsolatedAsyncioTestCase await processor._push_turn_text(USER_TURN_INCOMPLETE_LONG_MARKER) text_frames = [f for f in pushed_frames if isinstance(f, LLMTextFrame)] - self.assertEqual(len(text_frames), 1) - self.assertEqual(text_frames[0].text, USER_TURN_INCOMPLETE_LONG_MARKER) - self.assertTrue(text_frames[0].skip_tts) + self.assertEqual(len(text_frames), 0) + + marker_frames = [f for f in pushed_frames if isinstance(f, LLMMarkerFrame)] + self.assertEqual(len(marker_frames), 1) + self.assertEqual(marker_frames[0].marker, USER_TURN_INCOMPLETE_LONG_MARKER) + self.assertTrue(marker_frames[0].append_to_context_immediately) completed = [f for f in pushed_frames if isinstance(f, UserTurnCompletedFrame)] self.assertEqual(len(completed), 0) @@ -123,10 +134,12 @@ class TestUserUserTurnCompletionLLMServiceMixin(unittest.IsolatedAsyncioTestCase # Now send the complete marker await processor._push_turn_text(f" {USER_TURN_COMPLETE_MARKER} How are you?") - # Two LLMTextFrames pushed (marker + content) plus the - # UserTurnCompletedFrame broadcast. + # One LLMTextFrame for the spoken portion; one LLMMarkerFrame for + # the marker; UserTurnCompletedFrame broadcast in both directions. text_frames = [f for f in pushed_frames if isinstance(f, LLMTextFrame)] - self.assertEqual(len(text_frames), 2) + self.assertEqual(len(text_frames), 1) + marker_frames = [f for f in pushed_frames if isinstance(f, LLMMarkerFrame)] + self.assertEqual(len(marker_frames), 1) async def test_turn_state_reset_after_llm_full_response_end_frame(self): """Test that _turn_complete_found is reset when LLMFullResponseEndFrame is pushed."""