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

Route turn-completion markers through LLMMarkerFrame

Browse Source 
Add an `LLMMarkerFrame(DataFrame)` for sideband LLM markers that need
to be persisted to context but should not flow through the standard
text path (TTS, transcript). The frame carries an
`append_to_context_immediately` flag so the assistant aggregator can
either commit the marker as a stand-alone message (○ / ◐) or merge it
with the upcoming aggregation as a prefix on the response (✓).

`UserTurnCompletionLLMServiceMixin` now emits `LLMMarkerFrame` instead
of pushing the marker as `LLMTextFrame(skip_tts=True)`, which fixes
the case where an incomplete-turn marker (○ / ◐) was aggregated by
the assistant aggregator but never committed to the context because
the assistant turn lifecycle didn't run to completion (no spoken
response, no `LLMFullResponseEndFrame`-driven `push_aggregation`).

The frame is intentionally generic so other components — STT services
with built-in turn signals, end-of-turn classifiers, custom
annotations — can use the same mechanism to inject sideband signals
into the assistant context.
This commit is contained in:
Aleix Conchillo Flaqué
2026-05-05 16:30:37 -07:00
parent 1fa0310ea8
commit d1c8162b0c
4 changed files with 111 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
src/pipecat/frames/frames.py
View File

@@ -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 "✓ <response>").
"""
marker: str
append_to_context_immediately: bool = True
@dataclass
class AggregatedTextFrame(TextFrame):
"""Text frame representing an aggregation of TextFrames.

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

@@ -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 "✓ <response>" 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

26
src/pipecat/turns/user_turn_completion_mixin.py
View File

@@ -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 "✓ <response>".
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:

57
tests/test_user_turn_completion_mixin.py
View File

@@ -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."""