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

Merge pull request #3316 from pipecat-ai/aleix/llm-user-aggreagtor-enable-interruptions

Browse Source 
turns(user): add support for enabling/disabling interruptions
This commit is contained in:
Aleix Conchillo Flaqué
2025-12-29 17:45:56 -08:00
committed by GitHub
parent e22a6c9e4d 9fe99ed880
commit ad4c22cf44
15 changed files with 228 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
changelog/3297.deprecated.md
View File

@@ -1 +1,12 @@
- `PipelineParams.allow_interruptions` is now deprecated, use `LLMUserAggregator`'s new parameter `user_mute_strategies` instead.
- `PipelineParams.allow_interruptions` is now deprecated, use `LLMUserAggregator`'s new parameter `turn_start_strategies` instead. For example, to disable interruptions but still get user turns you can do:
```python
context_aggregator = LLMContextAggregatorPair(
context,
user_params=LLMUserAggregatorParams(
turn_start_strategies=TurnStartStrategies(
user=[TranscriptionUserTurnStartStrategy(enable_interruptions=False)],
),
),
)
```

1
changelog/3316.added.md Normal file
View File

@@ -0,0 +1 @@
- Added `enable_interruptions` constructor argument to all user turn strategies. This tells the `LLMUserAggregator` to push or not push an `InterruptionFrame`.

1
changelog/3316.other.md Normal file
View File

@@ -0,0 +1 @@
- Added `52-live-transcription.py` foundational example demonstrating live transcription and translation from English to Spanish. In this example, the bot is not interruptible: as the user continues speaking, English transcriptions are queued, and the bot continuously translates and speaks each queued sentence in Spanish without being interrupted by new user speech.

140
examples/foundational/52-live-translation.py Normal file
View File

@@ -0,0 +1,140 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import os
from dotenv import load_dotenv
from loguru import logger
from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.audio.vad.vad_analyzer import VADParams
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
LLMContextAggregatorPair,
LLMUserAggregatorParams,
)
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.services.deepgram.stt import DeepgramSTTService
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.bot import TurnAnalyzerBotTurnStartStrategy
from pipecat.turns.turn_start_strategies import TurnStartStrategies
from pipecat.turns.user import TranscriptionUserTurnStartStrategy
load_dotenv(override=True)
# We store functions so objects (e.g. SileroVADAnalyzer) don't get
# instantiated. The function will be called when the desired transport gets
# selected.
transport_params = {
"daily": lambda: DailyParams(
audio_in_enabled=True,
audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
),
"twilio": lambda: FastAPIWebsocketParams(
audio_in_enabled=True,
audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
),
"webrtc": lambda: TransportParams(
audio_in_enabled=True,
audio_out_enabled=True,
vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
),
}
async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
logger.info(f"Starting bot")
stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
tts = CartesiaTTSService(
api_key=os.getenv("CARTESIA_API_KEY"),
voice_id="d4db5fb9-f44b-4bd1-85fa-192e0f0d75f9", # Spanish-speaking Lady
)
llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
messages = [
{
"role": "system",
"content": "You are a live translation assistant. Your sole purpose is to translate English text into Spanish. When you receive English text from the user, immediately translate it into natural, fluent Spanish. Do not add explanations, commentary, or extra information—only provide the Spanish translation of the text you receive.",
},
]
context = LLMContext(messages)
# We use the TranscriptionUserTurnStartStrategy to start a new user turn
# every time a transcription is received. We disable interruptions, so the
# user can continue speaking while the bot is transcribing, without
# interrupting the bot.
context_aggregator = LLMContextAggregatorPair(
context,
user_params=LLMUserAggregatorParams(
turn_start_strategies=TurnStartStrategies(
user=[TranscriptionUserTurnStartStrategy(enable_interruptions=False)],
bot=[TurnAnalyzerBotTurnStartStrategy(turn_analyzer=LocalSmartTurnAnalyzerV3())],
),
),
)
pipeline = Pipeline(
[
transport.input(), # Transport user input
stt, # STT
context_aggregator.user(), # User responses
llm, # LLM
tts, # TTS (bot will speak the chosen language)
transport.output(), # Transport bot output
context_aggregator.assistant(), # Assistant spoken responses
]
)
task = PipelineTask(
pipeline,
params=PipelineParams(
enable_metrics=True,
enable_usage_metrics=True,
),
idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
)
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
logger.info(f"Client connected")
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
logger.info(f"Client disconnected")
await task.cancel()
runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
await runner.run(task)
async def bot(runner_args: RunnerArguments):
"""Main bot entry point compatible with Pipecat Cloud."""
transport = await create_transport(runner_args, transport_params)
await run_bot(transport, runner_args)
if __name__ == "__main__":
from pipecat.runner.run import main
main()

2
src/pipecat/pipeline/task.py
View File

@@ -107,7 +107,7 @@ class PipelineParams(BaseModel):
allow_interruptions: Whether to allow pipeline interruptions.
.. deprecated:: 0.0.99
Use `LLMUserAggregator`'s new `user_mute_strategies` parameter instead.
Use `LLMUserAggregator`'s new `turn_start_strategies` parameter instead.
audio_in_sample_rate: Input audio sample rate in Hz.
audio_out_sample_rate: Output audio sample rate in Hz.

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

@@ -561,8 +561,11 @@ class LLMUserAggregator(LLMContextAggregator):
await s.reset()
if params.enable_user_speaking_frames:
# TODO(aleix): These frames should really come from the top of the pipeline.
# TODO(aleix): This frame should really come from the top of the pipeline.
await self.broadcast_frame(UserStartedSpeakingFrame)
if params.enable_interruptions:
# TODO(aleix): This frame should really come from the top of the pipeline.
await self.broadcast_frame(InterruptionFrame)
await self._call_event_handler("on_user_turn_started", strategy)

5
src/pipecat/turns/bot/external_bot_turn_start_strategy.py
View File

@@ -29,14 +29,15 @@ class ExternalBotTurnStartStrategy(BaseBotTurnStartStrategy):
"""
def __init__(self, *, timeout: float = 0.5):
def __init__(self, *, timeout: float = 0.5, **kwargs):
"""Initialize the external bot turn start strategy.
Args:
timeout: A short delay used internally to handle consecutive or
slightly delayed transcriptions.
**kwargs: Additional keyword arguments.
"""
super().__init__(enable_user_speaking_frames=False)
super().__init__(enable_user_speaking_frames=False, **kwargs)
self._timeout = timeout
self._text = ""
self._user_speaking = False

5
src/pipecat/turns/bot/transcription_bot_turn_start_strategy.py
View File

@@ -28,14 +28,15 @@ class TranscriptionBotTurnStartStrategy(BaseBotTurnStartStrategy):
multiple or delayed transcription frames gracefully.
"""
def __init__(self, *, timeout: float = 0.5):
def __init__(self, *, timeout: float = 0.5, **kwargs):
"""Initialize the transcription-based bot turn start strategy.
Args:
timeout: A short delay used internally to handle consecutive or
slightly delayed transcriptions.
**kwargs: Additional keyword arguments.
"""
super().__init__()
super().__init__(**kwargs)
self._timeout = timeout
self._text = ""
self._vad_user_speaking = False

5
src/pipecat/turns/bot/turn_analyzer_bot_turn_start_strategy.py
View File

@@ -35,14 +35,15 @@ class TurnAnalyzerBotTurnStartStrategy(BaseBotTurnStartStrategy):
"""
def __init__(self, *, turn_analyzer: BaseTurnAnalyzer, timeout: float = 0.5):
def __init__(self, *, turn_analyzer: BaseTurnAnalyzer, timeout: float = 0.5, **kwargs):
"""Initialize the bot turn start strategy.
Args:
turn_analyzer: The turn detection analyzer instance to detect end of user turn.
timeout: Short delay used internally to handle frame timing and event triggering.
**kwargs: Additional keyword arguments.
"""
super().__init__()
super().__init__(**kwargs)
self._turn_analyzer = turn_analyzer
self._timeout = timeout
self._text = ""

27
src/pipecat/turns/user/base_user_turn_start_strategy.py
View File

@@ -32,6 +32,7 @@ class UserTurnStartedParams:
"""
enable_interruptions: bool
enable_user_speaking_frames: bool
@@ -49,18 +50,27 @@ class BaseUserTurnStartStrategy(BaseObject):
- `on_user_turn_started`: Signals that a user turn has started.
"""
def __init__(self, *, enable_user_speaking_frames: bool = True, **kwargs):
def __init__(
self,
*,
enable_interruptions: bool = True,
enable_user_speaking_frames: bool = True,
**kwargs,
):
"""Initialize the base user turn start strategy.
Args:
enable_user_speaking_frames: If True, the aggregator will emit frames
indicating when the user starts speaking, as well as interruption
frames. This is enabled by default, but you may want to disable it
if another component (e.g., an STT service) is already generating
these frames.
enable_interruptions: If True, the user aggregator will emit an
interruption frame when the user turn starts.
enable_user_speaking_frames: If True, the user aggregator will emit
frames indicating when the user starts speaking, as well as
interruption frames. This is enabled by default, but you may want
to disable it if another component (e.g., an STT service) is
already generating these frames.
**kwargs: Additional keyword arguments.
"""
super().__init__(**kwargs)
self._enable_interruptions = enable_interruptions
self._enable_user_speaking_frames = enable_user_speaking_frames
self._task_manager: Optional[BaseTaskManager] = None
self._register_event_handler("on_push_frame", sync=True)
@@ -123,5 +133,8 @@ class BaseUserTurnStartStrategy(BaseObject):
"""Trigger the `on_user_turn_started` event."""
await self._call_event_handler(
"on_user_turn_started",
UserTurnStartedParams(enable_user_speaking_frames=self._enable_user_speaking_frames),
UserTurnStartedParams(
enable_interruptions=self._enable_interruptions,
enable_user_speaking_frames=self._enable_user_speaking_frames,
),
)

10
src/pipecat/turns/user/external_user_turn_start_strategy.py
View File

@@ -19,9 +19,13 @@ class ExternalUserTurnStartStrategy(BaseUserTurnStartStrategy):
"""
def __init__(self):
"""Initialize the external user turn start strategy."""
super().__init__(enable_user_speaking_frames=False)
def __init__(self, **kwargs):
"""Initialize the external user turn start strategy.
Args:
**kwargs: Additional keyword arguments.
"""
super().__init__(enable_user_speaking_frames=False, **kwargs)
async def process_frame(self, frame: Frame):
"""Process an incoming frame to detect user turn start.

5
src/pipecat/turns/user/min_words_user_turn_start_strategy.py
View File

@@ -27,7 +27,7 @@ class MinWordsUserTurnStartStrategy(BaseUserTurnStartStrategy):
"""
def __init__(self, *, min_words: int, use_interim: bool = True):
def __init__(self, *, min_words: int, use_interim: bool = True, **kwargs):
"""Initialize the minimum words bot turn start strategy.
Args:
@@ -35,8 +35,9 @@ class MinWordsUserTurnStartStrategy(BaseUserTurnStartStrategy):
start of a user turn.
use_interim: Whether to consider interim transcription frames for
earlier detection.
**kwargs: Additional keyword arguments.
"""
super().__init__()
super().__init__(**kwargs)
self._min_words = min_words
self._use_interim = use_interim
self._bot_speaking = False

26
src/pipecat/turns/user/transcription_user_turn_start_strategy.py
View File

@@ -6,7 +6,7 @@
"""User turn start strategy based on transcriptions."""
from pipecat.frames.frames import BotStartedSpeakingFrame, Frame, TranscriptionFrame
from pipecat.frames.frames import Frame, InterimTranscriptionFrame, TranscriptionFrame
from pipecat.turns.user.base_user_turn_start_strategy import BaseUserTurnStartStrategy
@@ -20,15 +20,10 @@ class TranscriptionUserTurnStartStrategy(BaseUserTurnStartStrategy):
"""
def __init__(self):
def __init__(self, *, use_interim: bool = True, **kwargs):
"""Initialize transcription-based user turn start strategy."""
super().__init__()
self._bot_speaking = False
async def reset(self):
"""Reset the strategy to its initial state."""
await super().reset()
self._bot_speaking = False
super().__init__(**kwargs)
self._use_interim = use_interim
async def process_frame(self, frame: Frame):
"""Process an incoming frame to detect the start of a user turn.
@@ -38,14 +33,7 @@ class TranscriptionUserTurnStartStrategy(BaseUserTurnStartStrategy):
"""
await super().process_frame(frame)
if isinstance(frame, BotStartedSpeakingFrame):
await self._handle_bot_started_speaking(frame)
elif isinstance(frame, TranscriptionFrame):
await self._handle_transcription(frame)
async def _handle_bot_started_speaking(self, _: BotStartedSpeakingFrame):
self._bot_speaking = True
async def _handle_transcription(self, _: TranscriptionFrame):
if self._bot_speaking:
if isinstance(frame, InterimTranscriptionFrame) and self._use_interim:
await self.trigger_user_turn_started()
elif isinstance(frame, TranscriptionFrame):
await self.trigger_user_turn_started()

28
tests/test_bot_turn_start_strategy.py
View File

@@ -33,7 +33,7 @@ class TestTranscriptionBotTurnStartStrategy(unittest.IsolatedAsyncioTestCase):
should_start = None
@strategy.event_handler("on_bot_turn_started")
async def on_bot_turn_started(strategy, enable_user_speaking_frames):
async def on_bot_turn_started(strategy, params):
nonlocal should_start
should_start = True
@@ -58,7 +58,7 @@ class TestTranscriptionBotTurnStartStrategy(unittest.IsolatedAsyncioTestCase):
should_start = None
@strategy.event_handler("on_bot_turn_started")
async def on_bot_turn_started(strategy, enable_user_speaking_frames):
async def on_bot_turn_started(strategy, params):
nonlocal should_start
should_start = True
@@ -89,7 +89,7 @@ class TestTranscriptionBotTurnStartStrategy(unittest.IsolatedAsyncioTestCase):
should_start = None
@strategy.event_handler("on_bot_turn_started")
async def on_bot_turn_started(strategy, enable_user_speaking_frames):
async def on_bot_turn_started(strategy, params):
nonlocal should_start
should_start = True
@@ -136,7 +136,7 @@ class TestTranscriptionBotTurnStartStrategy(unittest.IsolatedAsyncioTestCase):
should_start = None
@strategy.event_handler("on_bot_turn_started")
async def on_bot_turn_started(strategy, enable_user_speaking_frames):
async def on_bot_turn_started(strategy, params):
nonlocal should_start
should_start = True
@@ -170,7 +170,7 @@ class TestTranscriptionBotTurnStartStrategy(unittest.IsolatedAsyncioTestCase):
should_start = None
@strategy.event_handler("on_bot_turn_started")
async def on_bot_turn_started(strategy, enable_user_speaking_frames):
async def on_bot_turn_started(strategy, params):
nonlocal should_start
should_start = True
@@ -212,7 +212,7 @@ class TestTranscriptionBotTurnStartStrategy(unittest.IsolatedAsyncioTestCase):
should_start = None
@strategy.event_handler("on_bot_turn_started")
async def on_bot_turn_started(strategy, enable_user_speaking_frames):
async def on_bot_turn_started(strategy, params):
nonlocal should_start
should_start = True
@@ -242,7 +242,7 @@ class TestTranscriptionBotTurnStartStrategy(unittest.IsolatedAsyncioTestCase):
should_start = None
@strategy.event_handler("on_bot_turn_started")
async def on_bot_turn_started(strategy, enable_user_speaking_frames):
async def on_bot_turn_started(strategy, params):
nonlocal should_start
should_start = True
@@ -278,7 +278,7 @@ class TestTranscriptionBotTurnStartStrategy(unittest.IsolatedAsyncioTestCase):
should_start = None
@strategy.event_handler("on_bot_turn_started")
async def on_bot_turn_started(strategy, enable_user_speaking_frames):
async def on_bot_turn_started(strategy, params):
nonlocal should_start
should_start = True
@@ -316,7 +316,7 @@ class TestTranscriptionBotTurnStartStrategy(unittest.IsolatedAsyncioTestCase):
should_start = None
@strategy.event_handler("on_bot_turn_started")
async def on_bot_turn_started(strategy, enable_user_speaking_frames):
async def on_bot_turn_started(strategy, params):
nonlocal should_start
should_start = True
@@ -350,7 +350,7 @@ class TestTranscriptionBotTurnStartStrategy(unittest.IsolatedAsyncioTestCase):
should_start = None
@strategy.event_handler("on_bot_turn_started")
async def on_bot_turn_started(strategy, enable_user_speaking_frames):
async def on_bot_turn_started(strategy, params):
nonlocal should_start
should_start = True
@@ -395,7 +395,7 @@ class TestTranscriptionBotTurnStartStrategy(unittest.IsolatedAsyncioTestCase):
should_start = None
@strategy.event_handler("on_bot_turn_started")
async def on_bot_turn_started(strategy, enable_user_speaking_frames):
async def on_bot_turn_started(strategy, params):
nonlocal should_start
should_start = True
@@ -415,7 +415,7 @@ class TestTranscriptionBotTurnStartStrategy(unittest.IsolatedAsyncioTestCase):
should_start = None
@strategy.event_handler("on_bot_turn_started")
async def on_bot_turn_started(strategy, enable_user_speaking_frames):
async def on_bot_turn_started(strategy, params):
nonlocal should_start
should_start = True
@@ -440,7 +440,7 @@ class TestTranscriptionBotTurnStartStrategy(unittest.IsolatedAsyncioTestCase):
should_start = None
@strategy.event_handler("on_bot_turn_started")
async def on_bot_turn_started(strategy, enable_user_speaking_frames):
async def on_bot_turn_started(strategy, params):
nonlocal should_start
should_start = True
@@ -484,7 +484,7 @@ class TestExternalBotTurnStartStrategy(unittest.IsolatedAsyncioTestCase):
should_start = None
@strategy.event_handler("on_bot_turn_started")
async def on_bot_turn_started(strategy, enable_user_speaking_frames):
async def on_bot_turn_started(strategy, params):
nonlocal should_start
should_start = True

21
tests/test_user_turn_start_strategy.py
View File

@@ -29,7 +29,7 @@ class TestMinWordsInterruptionStrategy(unittest.IsolatedAsyncioTestCase):
should_start = None
@strategy.event_handler("on_user_turn_started")
async def on_user_turn_started(strategy, enable_user_speaking_frames):
async def on_user_turn_started(strategy, params):
nonlocal should_start
should_start = True
@@ -61,7 +61,7 @@ class TestMinWordsInterruptionStrategy(unittest.IsolatedAsyncioTestCase):
should_start = None
@strategy.event_handler("on_user_turn_started")
async def on_user_turn_started(strategy, enable_user_speaking_frames):
async def on_user_turn_started(strategy, params):
nonlocal should_start
should_start = True
@@ -83,7 +83,7 @@ class TestMinWordsInterruptionStrategy(unittest.IsolatedAsyncioTestCase):
should_start = None
@strategy.event_handler("on_user_turn_started")
async def on_user_turn_started(strategy, enable_user_speaking_frames):
async def on_user_turn_started(strategy, params):
nonlocal should_start
should_start = True
@@ -104,7 +104,7 @@ class TestMinWordsInterruptionStrategy(unittest.IsolatedAsyncioTestCase):
should_start = None
@strategy.event_handler("on_user_turn_started")
async def on_user_turn_started(strategy, enable_user_speaking_frames):
async def on_user_turn_started(strategy, params):
nonlocal should_start
should_start = True
@@ -117,7 +117,7 @@ class TestMinWordsInterruptionStrategy(unittest.IsolatedAsyncioTestCase):
should_start = None
@strategy.event_handler("on_user_turn_started")
async def on_user_turn_started(strategy, enable_user_speaking_frames):
async def on_user_turn_started(strategy, params):
nonlocal should_start
should_start = True
@@ -134,7 +134,7 @@ class TestVADUserTurnStartStrategy(unittest.IsolatedAsyncioTestCase):
should_start = None
@strategy.event_handler("on_user_turn_started")
async def on_user_turn_started(strategy, enable_user_speaking_frames):
async def on_user_turn_started(strategy, params):
nonlocal should_start
should_start = True
@@ -152,14 +152,11 @@ class TestTranscriptionUserTurnStartStrategy(unittest.IsolatedAsyncioTestCase):
should_start = None
@strategy.event_handler("on_user_turn_started")
async def on_user_turn_started(strategy, enable_user_speaking_frames):
async def on_user_turn_started(strategy, params):
nonlocal should_start
should_start = True
await strategy.process_frame(TranscriptionFrame(text="Hello!", user_id="", timestamp="now"))
self.assertFalse(should_start)
await strategy.process_frame(BotStartedSpeakingFrame())
await strategy.process_frame(VADUserStartedSpeakingFrame())
self.assertFalse(should_start)
await strategy.process_frame(TranscriptionFrame(text="Hello!", user_id="", timestamp="now"))
@@ -173,7 +170,7 @@ class TestExternalUserTurnStartStrategy(unittest.IsolatedAsyncioTestCase):
should_start = None
@strategy.event_handler("on_user_turn_started")
async def on_user_turn_started(strategy, enable_user_speaking_frames):
async def on_user_turn_started(strategy, params):
nonlocal should_start
should_start = True