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

Split user-turn-stop into inference-triggered and finalized events

Browse Source 
Fixes a real bug: with `filter_incomplete_user_turns` enabled, the
smart-turn detector's tentative stop was firing `on_user_turn_stopped`
before the LLM had a chance to veto it. Observers, transcript
appenders and UI indicators received an early — and sometimes
duplicated — signal.

Decomposes the single stop concern into two events:
- `on_user_turn_inference_triggered` fires when a stop strategy has
  enough signal to start LLM inference. The aggregator pushes the
  context here, kicking off the LLM call.
- `on_user_turn_stopped` fires only when the user turn is semantically
  final. Built-in strategies fire both events at the same call site,
  preserving today's behavior for the common case.

Adds `LLMTurnCompletionUserTurnStopStrategy`, which gates
finalization on a `UserTurnCompletedFrame` (a fieldless system frame
emitted by any component judging turn completeness — currently the
`UserTurnCompletionLLMServiceMixin` on `✓`).

Adds `deferred(strategy)` / `DeferredUserTurnStopStrategy`, a thin
wrapper that forwards an inner strategy's events except
`on_user_turn_stopped`. Use this to install a stop strategy as an
inference trigger only, leaving finalization to a peer (e.g. the LLM
completion strategy).

Adds `llm_completion_user_turn_stop_strategies()` for the common
case:

    UserTurnStrategies(
        stop=llm_completion_user_turn_stop_strategies(),
    )

Deprecates `LLMUserAggregatorParams.filter_incomplete_user_turns`.
The aggregator emits a `DeprecationWarning`, wraps existing stop
strategies with `deferred(...)`, and appends
`LLMTurnCompletionUserTurnStopStrategy` automatically.
This commit is contained in:
Aleix Conchillo Flaqué
2026-05-01 14:53:53 -07:00
parent 1073510574
commit 480eca42f5
14 changed files with 712 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

138
tests/test_context_aggregators_universal.py
View File

@@ -179,8 +179,16 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase):
assert context.messages[0]["content"] == "Hi there!"
async def test_llm_messages_update_does_not_inject_turn_completion_into_context(self):
from pipecat.turns.user_turn_strategies import (
llm_completion_user_turn_stop_strategies,
)
context = LLMContext()
params = LLMUserAggregatorParams(filter_incomplete_user_turns=True)
params = LLMUserAggregatorParams(
user_turn_strategies=UserTurnStrategies(
stop=llm_completion_user_turn_stop_strategies(),
),
)
pipeline = Pipeline([LLMUserAggregator(context, params=params)])
new_messages = [
@@ -291,8 +299,8 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase):
UserStartedSpeakingFrame,
InterruptionFrame,
VADUserStoppedSpeakingFrame,
UserStoppedSpeakingFrame,
LLMContextFrame,
UserStoppedSpeakingFrame,
]
await run_test(
pipeline,
@@ -563,6 +571,132 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase):
expected_down_frames=[SpeechControlParamsFrame],
)
async def test_inference_triggered_event_fires_on_default_strategies(self):
"""Default flow fires inference-triggered before stopped, both with the same strategy."""
from pipecat.frames.frames import UserTurnCompletedFrame # noqa: F401
context = LLMContext()
user_aggregator = LLMUserAggregator(
context,
params=LLMUserAggregatorParams(
user_turn_strategies=UserTurnStrategies(
stop=[
SpeechTimeoutUserTurnStopStrategy(user_speech_timeout=TRANSCRIPTION_TIMEOUT)
]
),
),
)
events: list[str] = []
@user_aggregator.event_handler("on_user_turn_inference_triggered")
async def on_inference_triggered(aggregator, strategy):
events.append("inference_triggered")
@user_aggregator.event_handler("on_user_turn_stopped")
async def on_stopped(aggregator, strategy, message):
events.append(f"stopped:{message.content}")
pipeline = Pipeline([user_aggregator])
frames_to_send = [
VADUserStartedSpeakingFrame(),
TranscriptionFrame(text="Hi!", user_id="", timestamp="now"),
SleepFrame(),
VADUserStoppedSpeakingFrame(),
SleepFrame(sleep=TRANSCRIPTION_TIMEOUT + 0.1),
]
await run_test(pipeline, frames_to_send=frames_to_send)
self.assertEqual(events, ["inference_triggered", "stopped:Hi!"])
async def test_filter_incomplete_user_turns_emits_deprecation_warning(self):
"""Setting the legacy flag emits a DeprecationWarning."""
import warnings
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter("always")
LLMUserAggregatorParams(filter_incomplete_user_turns=True)
matched = [
x
for x in w
if issubclass(x.category, DeprecationWarning)
and "filter_incomplete_user_turns" in str(x.message)
]
self.assertTrue(matched, "expected a DeprecationWarning")
async def test_filter_incomplete_user_turns_installs_strategy(self):
"""Legacy flag wraps existing stops with deferred() and appends the LLM strategy."""
import warnings
from pipecat.turns.user_stop import (
DeferredUserTurnStopStrategy,
LLMTurnCompletionUserTurnStopStrategy,
SpeechTimeoutUserTurnStopStrategy,
)
existing = SpeechTimeoutUserTurnStopStrategy(user_speech_timeout=TRANSCRIPTION_TIMEOUT)
context = LLMContext()
with warnings.catch_warnings():
warnings.simplefilter("ignore", DeprecationWarning)
params = LLMUserAggregatorParams(
filter_incomplete_user_turns=True,
user_turn_strategies=UserTurnStrategies(stop=[existing]),
)
aggregator = LLMUserAggregator(context, params=params)
stop_strategies = aggregator._params.user_turn_strategies.stop
self.assertEqual(len(stop_strategies), 2)
self.assertIsInstance(stop_strategies[0], DeferredUserTurnStopStrategy)
self.assertIs(stop_strategies[0].inner, existing)
self.assertIsInstance(stop_strategies[1], LLMTurnCompletionUserTurnStopStrategy)
async def test_llm_completion_strategy_finalizes_on_complete_marker(self):
"""LLMTurnCompletionUserTurnStopStrategy finalizes only on UserTurnCompletedFrame(complete)."""
from pipecat.frames.frames import UserTurnCompletedFrame
from pipecat.turns.user_stop import LLMTurnCompletionUserTurnStopStrategy, deferred
gating = LLMTurnCompletionUserTurnStopStrategy()
upstream = deferred(
SpeechTimeoutUserTurnStopStrategy(user_speech_timeout=TRANSCRIPTION_TIMEOUT)
)
context = LLMContext()
user_aggregator = LLMUserAggregator(
context,
params=LLMUserAggregatorParams(
user_turn_strategies=UserTurnStrategies(stop=[upstream, gating]),
),
)
events: list[str] = []
@user_aggregator.event_handler("on_user_turn_inference_triggered")
async def on_inference_triggered(aggregator, strategy):
events.append("inference_triggered")
@user_aggregator.event_handler("on_user_turn_stopped")
async def on_stopped(aggregator, strategy, message):
events.append("stopped")
pipeline = Pipeline([user_aggregator])
# Drive the pipeline. Inference fires after the upstream
# strategy's timeout. Stop fires only when UserTurnCompletedFrame
# arrives (producer absence == "not yet complete").
frames_to_send = [
VADUserStartedSpeakingFrame(),
TranscriptionFrame(text="Hi", user_id="", timestamp="now"),
SleepFrame(),
VADUserStoppedSpeakingFrame(),
SleepFrame(sleep=TRANSCRIPTION_TIMEOUT + 0.1),
# At this point inference_triggered should have fired but NOT stopped.
UserTurnCompletedFrame(),
SleepFrame(),
]
await run_test(pipeline, frames_to_send=frames_to_send)
self.assertEqual(events, ["inference_triggered", "stopped"])
class TestLLMAssistantAggregator(unittest.IsolatedAsyncioTestCase):
async def test_empty(self):

61
tests/test_user_turn_completion_mixin.py
View File

@@ -8,7 +8,7 @@ import unittest
import unittest.mock
from unittest.mock import AsyncMock
from pipecat.frames.frames import LLMFullResponseEndFrame, LLMTextFrame
from pipecat.frames.frames import LLMFullResponseEndFrame, LLMTextFrame, UserTurnCompletedFrame
from pipecat.processors.frame_processor import FrameProcessor
from pipecat.services.llm_service import LLMService
from pipecat.services.settings import LLMSettings
@@ -44,21 +44,25 @@ class TestUserUserTurnCompletionLLMServiceMixin(unittest.IsolatedAsyncioTestCase
# Simulate LLM generating: "✓ Hello there!"
await processor._push_turn_text(f"{USER_TURN_COMPLETE_MARKER} Hello there!")
# Should have 2 text frames: marker (skip_tts) and content (normal)
self.assertEqual(len(pushed_frames), 2)
# Two LLMTextFrames: marker (skip_tts) and content (normal). The
# broadcast also pushes UserTurnCompletedFrame upstream + downstream.
text_frames = [f for f in pushed_frames if isinstance(f, LLMTextFrame)]
self.assertEqual(len(text_frames), 2)
# First frame should be the marker with skip_tts=True
self.assertIsInstance(pushed_frames[0], LLMTextFrame)
self.assertEqual(pushed_frames[0].text, USER_TURN_COMPLETE_MARKER)
self.assertTrue(pushed_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 frame should be the actual text without skip_tts
self.assertIsInstance(pushed_frames[1], LLMTextFrame)
self.assertEqual(pushed_frames[1].text, "Hello there!")
self.assertFalse(pushed_frames[1].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)
# 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."""
"""Test that ○ marker suppresses text with skip_tts and emits no completed frame."""
processor = MockProcessor()
pushed_frames = []
@@ -70,14 +74,17 @@ class TestUserUserTurnCompletionLLMServiceMixin(unittest.IsolatedAsyncioTestCase
await processor._push_turn_text(USER_TURN_INCOMPLETE_SHORT_MARKER)
# Should have 1 text frame with skip_tts=True
self.assertEqual(len(pushed_frames), 1)
self.assertIsInstance(pushed_frames[0], LLMTextFrame)
self.assertEqual(pushed_frames[0].text, USER_TURN_INCOMPLETE_SHORT_MARKER)
self.assertTrue(pushed_frames[0].skip_tts)
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)
# 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."""
"""Test that ◐ marker suppresses text with skip_tts and emits no completed frame."""
processor = MockProcessor()
pushed_frames = []
@@ -89,11 +96,13 @@ class TestUserUserTurnCompletionLLMServiceMixin(unittest.IsolatedAsyncioTestCase
await processor._push_turn_text(USER_TURN_INCOMPLETE_LONG_MARKER)
# Should have 1 text frame with skip_tts=True
self.assertEqual(len(pushed_frames), 1)
self.assertIsInstance(pushed_frames[0], LLMTextFrame)
self.assertEqual(pushed_frames[0].text, USER_TURN_INCOMPLETE_LONG_MARKER)
self.assertTrue(pushed_frames[0].skip_tts)
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)
completed = [f for f in pushed_frames if isinstance(f, UserTurnCompletedFrame)]
self.assertEqual(len(completed), 0)
async def test_text_buffered_until_marker_found(self):
"""Test that text is buffered until a marker is detected."""
@@ -114,8 +123,10 @@ class TestUserUserTurnCompletionLLMServiceMixin(unittest.IsolatedAsyncioTestCase
# Now send the complete marker
await processor._push_turn_text(f" {USER_TURN_COMPLETE_MARKER} How are you?")
# Now frames should be pushed
self.assertEqual(len(pushed_frames), 2)
# Two LLMTextFrames pushed (marker + content) plus the
# UserTurnCompletedFrame broadcast.
text_frames = [f for f in pushed_frames if isinstance(f, LLMTextFrame)]
self.assertEqual(len(text_frames), 2)
async def test_turn_state_reset_after_llm_full_response_end_frame(self):
"""Test that _turn_complete_found is reset when LLMFullResponseEndFrame is pushed."""

62
tests/test_user_turn_controller.py
View File

@@ -19,7 +19,7 @@ from pipecat.turns.user_start import VADUserTurnStartStrategy
from pipecat.turns.user_start.min_words_user_turn_start_strategy import (
MinWordsUserTurnStartStrategy,
)
from pipecat.turns.user_stop import SpeechTimeoutUserTurnStopStrategy
from pipecat.turns.user_stop import SpeechTimeoutUserTurnStopStrategy, deferred
from pipecat.turns.user_turn_controller import UserTurnController
from pipecat.turns.user_turn_strategies import ExternalUserTurnStrategies, UserTurnStrategies
from pipecat.utils.asyncio.task_manager import TaskManager, TaskManagerParams
@@ -71,6 +71,66 @@ class TestUserTurnController(unittest.IsolatedAsyncioTestCase):
await asyncio.sleep(TRANSCRIPTION_TIMEOUT + 0.1)
self.assertTrue(should_stop)
async def test_inference_triggered_fires_alongside_stopped(self):
"""Default strategies fire both inference-triggered and stopped, in order."""
controller = UserTurnController(
user_turn_strategies=UserTurnStrategies(
stop=[SpeechTimeoutUserTurnStopStrategy(user_speech_timeout=TRANSCRIPTION_TIMEOUT)],
)
)
await controller.setup(self.task_manager)
events: list[str] = []
@controller.event_handler("on_user_turn_inference_triggered")
async def on_user_turn_inference_triggered(controller, strategy):
events.append("inference_triggered")
@controller.event_handler("on_user_turn_stopped")
async def on_user_turn_stopped(controller, strategy, params):
events.append("stopped")
await controller.process_frame(VADUserStartedSpeakingFrame())
await controller.process_frame(
TranscriptionFrame(text="Hello!", user_id="", timestamp="now")
)
await controller.process_frame(VADUserStoppedSpeakingFrame())
await asyncio.sleep(TRANSCRIPTION_TIMEOUT + 0.1)
self.assertEqual(events, ["inference_triggered", "stopped"])
async def test_deferred_wrapper_skips_stopped(self):
"""A deferred() wrapper drops the inner strategy's on_user_turn_stopped event."""
wrapped = deferred(
SpeechTimeoutUserTurnStopStrategy(user_speech_timeout=TRANSCRIPTION_TIMEOUT)
)
controller = UserTurnController(user_turn_strategies=UserTurnStrategies(stop=[wrapped]))
await controller.setup(self.task_manager)
events: list[str] = []
@controller.event_handler("on_user_turn_inference_triggered")
async def on_user_turn_inference_triggered(controller, strategy):
events.append("inference_triggered")
@controller.event_handler("on_user_turn_stopped")
async def on_user_turn_stopped(controller, strategy, params):
events.append("stopped")
await controller.process_frame(VADUserStartedSpeakingFrame())
await controller.process_frame(
TranscriptionFrame(text="Hello!", user_id="", timestamp="now")
)
await controller.process_frame(VADUserStoppedSpeakingFrame())
await asyncio.sleep(TRANSCRIPTION_TIMEOUT + 0.1)
# The inner strategy fires inference-triggered (forwarded by the
# wrapper). Finalization is suppressed, but the controller's
# stop watchdog eventually fires `stopped`.
self.assertEqual(events[0], "inference_triggered")
async def test_user_turn_start_reset(self):
controller = UserTurnController(
user_turn_strategies=UserTurnStrategies(