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

Replace llm_completion_user_turn_stop_strategies() with FilterIncompleteUserTurnStrategies

Browse Source 
Wrap the detector chain with `deferred(...)` and append the LLM
completion gate via a `UserTurnStrategies` specialization rather than
a free-standing helper, mirroring the existing
`ExternalUserTurnStrategies` pattern. The class lives next to other
strategy containers in `pipecat.turns.user_turn_strategies`, so users
discover it where they're already configuring `user_turn_strategies`.

The deprecated `filter_incomplete_user_turns` flag now rewires
through `FilterIncompleteUserTurnStrategies` under the hood, keeping
the migration path identical to before. `deferred(...)` stays public
as the explicit escape hatch for non-default compositions.
This commit is contained in:
Aleix Conchillo Flaqué
2026-05-06 11:12:32 -07:00
parent e3e90d38aa
commit 952dddca8b
6 changed files with 94 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
changelog/4405.added.4.md
View File

@@ -1 +1 @@
- Added `llm_completion_user_turn_stop_strategies()` in `pipecat.turns.user_turn_strategies` — a one-call helper that pairs default (or user-supplied) stop strategies, each wrapped with `deferred(...)`, with `LLMTurnCompletionUserTurnStopStrategy`. Common case: `stop=llm_completion_user_turn_stop_strategies()`.
- Added `FilterIncompleteUserTurnStrategies` in `pipecat.turns.user_turn_strategies` — a `UserTurnStrategies` specialization that wraps the detector chain with `deferred(...)` and appends `LLMTurnCompletionUserTurnStopStrategy` as the finalizer. Common case: `user_turn_strategies=FilterIncompleteUserTurnStrategies()`. Pass `config=UserTurnCompletionConfig(...)` to customize timeouts and prompts.

2
changelog/4405.deprecated.md
View File

@@ -1 +1 @@
- Deprecated `LLMUserAggregatorParams.filter_incomplete_user_turns`. Add `LLMTurnCompletionUserTurnStopStrategy` to `user_turn_strategies.stop` instead (typically via `llm_completion_user_turn_stop_strategies()`). Setting the legacy flag still works for one release: the aggregator emits a `DeprecationWarning`, wraps existing stop strategies with `deferred(...)`, and appends the new finalizer strategy.
- Deprecated `LLMUserAggregatorParams.filter_incomplete_user_turns`. Use `user_turn_strategies=FilterIncompleteUserTurnStrategies()` (or add `LLMTurnCompletionUserTurnStopStrategy` to a custom `user_turn_strategies.stop`) instead. Setting the legacy flag still works for one release: the aggregator emits a `DeprecationWarning` and rewires the strategies as if you had passed `FilterIncompleteUserTurnStrategies` directly.

32
examples/turn-management/turn-management-filter-incomplete-turns.py
View File

@@ -41,10 +41,7 @@ from pipecat.services.openai.llm import OpenAILLMService
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.daily.transport import DailyParams
from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
from pipecat.turns.user_turn_strategies import (
UserTurnStrategies,
llm_completion_user_turn_stop_strategies,
)
from pipecat.turns.user_turn_strategies import FilterIncompleteUserTurnStrategies
load_dotenv(override=True)
@@ -87,11 +84,11 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
)
context = LLMContext()
# `llm_completion_user_turn_stop_strategies()` pairs the default
# stop strategies with `LLMTurnCompletionUserTurnStopStrategy`:
# those strategies trigger LLM inference but the public
# `on_user_turn_stopped` event fires only when the LLM confirms ✓.
# The LLM marks each response with one of:
# `FilterIncompleteUserTurnStrategies` pairs the default detector
# chain with `LLMTurnCompletionUserTurnStopStrategy`: detectors
# trigger LLM inference but the public `on_user_turn_stopped` event
# fires only when the LLM confirms ✓. The LLM marks each response
# with one of:
# ✓ = complete (respond normally)
# ○ = incomplete short (wait 5s, then prompt)
# ◐ = incomplete long (wait 15s, then prompt)
@@ -99,17 +96,14 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
context,
user_params=LLMUserAggregatorParams(
vad_analyzer=SileroVADAnalyzer(),
user_turn_strategies=UserTurnStrategies(
stop=llm_completion_user_turn_stop_strategies(),
user_turn_strategies=FilterIncompleteUserTurnStrategies(
# Optional: customize turn completion behavior
# stop=llm_completion_user_turn_stop_strategies(
# config=UserTurnCompletionConfig(
# incomplete_short_timeout=5.0,
# incomplete_long_timeout=15.0,
# incomplete_short_prompt="Custom prompt...",
# incomplete_long_prompt="Custom prompt...",
# instructions="Custom turn completion instructions...",
# ),
# config=UserTurnCompletionConfig(
# incomplete_short_timeout=5.0,
# incomplete_long_timeout=15.0,
# incomplete_short_prompt="Custom prompt...",
# incomplete_long_prompt="Custom prompt...",
# instructions="Custom turn completion instructions...",
# ),
),
),

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

@@ -85,7 +85,10 @@ from pipecat.turns.user_start import BaseUserTurnStartStrategy, UserTurnStartedP
from pipecat.turns.user_stop import BaseUserTurnStopStrategy, UserTurnStoppedParams
from pipecat.turns.user_turn_completion_mixin import UserTurnCompletionConfig
from pipecat.turns.user_turn_controller import UserTurnController
from pipecat.turns.user_turn_strategies import UserTurnStrategies
from pipecat.turns.user_turn_strategies import (
FilterIncompleteUserTurnStrategies,
UserTurnStrategies,
)
from pipecat.utils.context.llm_context_summarization import (
LLMAutoContextSummarizationConfig,
LLMContextSummarizationConfig,
@@ -123,18 +126,18 @@ class LLMUserAggregatorParams:
has been idle (not speaking) for this duration. Set to 0 to disable
idle detection.
vad_analyzer: Voice Activity Detection analyzer instance.
filter_incomplete_user_turns: [DEPRECATED] Add
:class:`~pipecat.turns.user_stop.LLMTurnCompletionUserTurnStopStrategy`
to ``user_turn_strategies.stop`` instead. When enabled, the LLM
outputs a turn-completion marker at the start of each response:
✓ (complete), ○ (incomplete short), or ◐ (incomplete long).
Incomplete responses are suppressed and timeouts trigger
re-prompting.
filter_incomplete_user_turns: [DEPRECATED] Use
``user_turn_strategies=FilterIncompleteUserTurnStrategies()``
instead. When enabled, the LLM outputs a turn-completion
marker at the start of each response: ✓ (complete), ○
(incomplete short), or ◐ (incomplete long). Incomplete
responses are suppressed and timeouts trigger re-prompting.
user_turn_completion_config: [DEPRECATED] Configuration for turn
completion behavior including custom instructions, timeouts, and
prompts. Only used when filter_incomplete_user_turns is True
(deprecated path) — for the new strategy-based API, pass the config
directly to ``LLMTurnCompletionUserTurnStopStrategy(config=...)``.
directly to ``FilterIncompleteUserTurnStrategies(config=...)``.
"""
add_tool_change_messages: bool = False
@@ -151,16 +154,14 @@ class LLMUserAggregatorParams:
if self.filter_incomplete_user_turns:
warnings.warn(
"LLMUserAggregatorParams.filter_incomplete_user_turns is deprecated. "
"Add LLMTurnCompletionUserTurnStopStrategy to "
"user_turn_strategies.stop instead.",
"Use user_turn_strategies=FilterIncompleteUserTurnStrategies() instead.",
DeprecationWarning,
stacklevel=2,
)
if self.user_turn_completion_config:
warnings.warn(
"LLMUserAggregatorParams.user_turn_completion_config is deprecated. "
"Add LLMTurnCompletionUserTurnStopStrategy to "
"user_turn_strategies.stop instead.",
"Use user_turn_strategies=FilterIncompleteUserTurnStrategies() instead.",
DeprecationWarning,
stacklevel=2,
)
@@ -570,21 +571,15 @@ class LLMUserAggregator(LLMContextAggregator):
user_turn_strategies = self._params.user_turn_strategies or UserTurnStrategies()
# Deprecated path: translate filter_incomplete_user_turns into
# wrapping pre-existing stop strategies with deferred() and
# appending LLMTurnCompletionUserTurnStopStrategy. The
# the equivalent FilterIncompleteUserTurnStrategies wiring. The
# DeprecationWarning is emitted in LLMUserAggregatorParams.__post_init__.
if self._params.filter_incomplete_user_turns:
from pipecat.turns.user_stop import (
LLMTurnCompletionUserTurnStopStrategy,
deferred,
user_turn_strategies = FilterIncompleteUserTurnStrategies(
start=user_turn_strategies.start,
stop=user_turn_strategies.stop,
config=self._params.user_turn_completion_config,
)
existing_stop = list(user_turn_strategies.stop or [])
user_turn_strategies.stop = [deferred(s) for s in existing_stop] + [
LLMTurnCompletionUserTurnStopStrategy(
config=self._params.user_turn_completion_config
)
]
self._params.user_turn_strategies = user_turn_strategies
self._user_is_muted = False
self._user_turn_start_timestamp = ""

90
src/pipecat/turns/user_turn_strategies.py
View File

@@ -51,42 +51,6 @@ def default_user_turn_stop_strategies() -> list[BaseUserTurnStopStrategy]:
return [TurnAnalyzerUserTurnStopStrategy(turn_analyzer=LocalSmartTurnAnalyzerV3())]
def llm_completion_user_turn_stop_strategies(
strategies: list[BaseUserTurnStopStrategy] | None = None,
*,
config: UserTurnCompletionConfig | None = None,
) -> list[BaseUserTurnStopStrategy]:
"""Build a stop-strategy list gated on the LLM's turn-completion verdict.
Wraps ``strategies`` with :func:`deferred` so they trigger inference
but don't fire ``on_user_turn_stopped`` themselves, then appends
:class:`~pipecat.turns.user_stop.LLMTurnCompletionUserTurnStopStrategy`
as the finalizer. Use as the ``stop`` field of a
:class:`UserTurnStrategies`::
UserTurnStrategies(
stop=llm_completion_user_turn_stop_strategies(),
)
Args:
strategies: Stop strategies that should drive inference. If
None, uses :func:`default_user_turn_stop_strategies`.
config: Optional configuration applied to the LLM via the
``filter_incomplete_user_turns`` setting. Customizes the
turn-completion instructions, incomplete-turn timeouts, and
re-prompts.
Returns:
``[deferred(s) for s in strategies] +
[LLMTurnCompletionUserTurnStopStrategy(config=config)]``.
"""
strategies = strategies if strategies is not None else default_user_turn_stop_strategies()
return [
*(deferred(s) for s in strategies),
LLMTurnCompletionUserTurnStopStrategy(config=config),
]
@dataclass
class UserTurnStrategies:
"""Container for user turn start and stop strategies.
@@ -134,3 +98,57 @@ class ExternalUserTurnStrategies(UserTurnStrategies):
def __post_init__(self):
self.start = [ExternalUserTurnStartStrategy()]
self.stop = [ExternalUserTurnStopStrategy()]
@dataclass
class FilterIncompleteUserTurnStrategies(UserTurnStrategies):
"""Stop strategies gated on the LLM's turn-completion verdict.
The LLM is asked to begin every response with one of three markers:
✓ (complete), ○ (incomplete short), or ◐ (incomplete long). Only ✓
finalizes the user turn; ○ / ◐ keep the turn open so the user can
continue speaking and the LLM can re-evaluate later.
Configuring strategies this way preserves the existing detector
chain (defaults or user-supplied) for inference triggering and
appends :class:`~pipecat.turns.user_stop.LLMTurnCompletionUserTurnStopStrategy`
as the finalizer. The detector strategies are wrapped with
:func:`~pipecat.turns.user_stop.deferred` automatically so they fire
only ``on_user_turn_inference_triggered`` and leave finalization to
the LLM gate.
Parameters:
config: Optional configuration applied to the LLM via the
``filter_incomplete_user_turns`` setting. Customizes the
turn-completion instructions, incomplete-turn timeouts, and
re-prompts. If None, defaults from
:class:`~pipecat.turns.user_turn_completion_mixin.UserTurnCompletionConfig`
are used.
Example::
user_turn_strategies=FilterIncompleteUserTurnStrategies()
# Custom detector chain:
user_turn_strategies=FilterIncompleteUserTurnStrategies(
stop=[SpeechTimeoutUserTurnStopStrategy(...)],
)
# Custom completion config:
user_turn_strategies=FilterIncompleteUserTurnStrategies(
config=UserTurnCompletionConfig(
incomplete_short_timeout=5.0,
incomplete_long_timeout=15.0,
),
)
"""
config: UserTurnCompletionConfig | None = None
def __post_init__(self):
super().__post_init__()
# Defer the detector chain so it only fires inference-triggered,
# then append the LLM gate as the sole finalizer.
gated: list[BaseUserTurnStopStrategy] = [deferred(s) for s in self.stop or []]
gated.append(LLMTurnCompletionUserTurnStopStrategy(config=self.config))
self.stop = gated

13
tests/test_context_aggregators_universal.py
View File

@@ -64,7 +64,10 @@ from pipecat.turns.user_mute import (
MuteUntilFirstBotCompleteUserMuteStrategy,
)
from pipecat.turns.user_stop import SpeechTimeoutUserTurnStopStrategy
from pipecat.turns.user_turn_strategies import UserTurnStrategies
from pipecat.turns.user_turn_strategies import (
FilterIncompleteUserTurnStrategies,
UserTurnStrategies,
)
from pipecat.utils.text.base_text_aggregator import AggregationType
USER_TURN_STOP_TIMEOUT = 0.2
@@ -179,15 +182,9 @@ 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(
user_turn_strategies=UserTurnStrategies(
stop=llm_completion_user_turn_stop_strategies(),
),
user_turn_strategies=FilterIncompleteUserTurnStrategies(),
)
pipeline = Pipeline([LLMUserAggregator(context, params=params)])