Align Together STT/TTS services with Pipecat patterns

STT:
- Add Settings class alias and 4-step init pattern
- Add resampler to convert pipeline audio to 16kHz for Together API
- Add keepalive support and _update_settings with reconnect
- Pass language to transcription frames
- Remove unnecessary OpenAI-Beta header

TTS:
- Add Settings class alias and 4-step init pattern
- Use push_start_frame=True for base class audio context management
- Route audio through append_to_audio_context instead of push_frame
- Track pending commits for proper audio context lifecycle
- Replace _handle_interruption with on_audio_context_interrupted
- Add _update_settings with reconnect
- Guard against stale audio after interruption

CHANGELOG.md

@@ -7,6 +7,218 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 <!-- towncrier release notes start -->
 
+## [0.0.106] - 2026-03-18
+
+### Added
+
+- Added optional `service` field to `ServiceUpdateSettingsFrame` (and its
+  subclasses `LLMUpdateSettingsFrame`, `TTSUpdateSettingsFrame`,
+  `STTUpdateSettingsFrame`) to target a specific service instance. When
+  `service` is set, only the matching service applies the settings; others
+  forward the frame unchanged. This enables updating a single service when
+  multiple services of the same type exist in the pipeline.
+  (PR [#4004](https://github.com/pipecat-ai/pipecat/pull/4004))
+
+- Added `sip_provider` and `room_geo` parameters to `configure()` in the Daily
+  runner. These convenience parameters let callers specify a SIP provider name
+  and geographic region directly without manually constructing
+  `DailyRoomProperties` and `DailyRoomSipParams`.
+  (PR [#4005](https://github.com/pipecat-ai/pipecat/pull/4005))
+
+- Added `PerplexityLLMAdapter` that automatically transforms conversation
+  messages to satisfy Perplexity's stricter API constraints (strict role
+  alternation, no non-initial system messages, last message must be user/tool).
+  Previously, certain conversation histories could cause Perplexity API errors
+  that didn't occur with OpenAI (`PerplexityLLMService` subclasses
+  `OpenAILLMService` since Perplexity uses an OpenAI-compatible API).
+  (PR [#4009](https://github.com/pipecat-ai/pipecat/pull/4009))
+
+- Added DTMF input event support to the Daily transport. Incoming DTMF tones
+  are now received via Daily's `on_dtmf_event` callback and pushed into the
+  pipeline as `InputDTMFFrame`, enabling bots to react to keypad presses from
+  phone callers.
+  (PR [#4047](https://github.com/pipecat-ai/pipecat/pull/4047))
+
+- Added `WakePhraseUserTurnStartStrategy` for triggering user turns based on
+  wake phrases, with support for `single_activation` mode. Deprecates
+  `WakeCheckFilter`.
+  (PR [#4064](https://github.com/pipecat-ai/pipecat/pull/4064))
+
+- Added `default_user_turn_start_strategies()` and
+  `default_user_turn_stop_strategies()` helper functions for composing custom
+  strategy lists.
+  (PR [#4064](https://github.com/pipecat-ai/pipecat/pull/4064))
+
+### Changed
+
+- Changed tool result JSON serialization to use `ensure_ascii=False`,
+  preserving UTF-8 characters instead of escaping them. This reduces context
+  size and token usage for non-English languages.
+  (PR [#3457](https://github.com/pipecat-ai/pipecat/pull/3457))
+
+- `OpenAIRealtimeSTTService`'s `noise_reduction` parameter is now part of
+  `OpenAIRealtimeSTTSettings`, making it runtime-updatable via
+  `STTUpdateSettingsFrame`. The direct `noise_reduction` init argument is
+  deprecated as of 0.0.106.
+  (PR [#3991](https://github.com/pipecat-ai/pipecat/pull/3991))
+
+- Updated `sarvamai` dependency from `0.1.26a2` (alpha) to `0.1.26` (stable
+  release).
+  (PR [#3997](https://github.com/pipecat-ai/pipecat/pull/3997))
+
+- `SimliVideoService` now extends `AIService` instead of `FrameProcessor`,
+  aligning it with the HeyGen and Tavus video services. It supports
+  `SimliVideoService.Settings(...)` for configuration and uses
+  `start()`/`stop()`/`cancel()` lifecycle methods. Existing constructor usage
+  (`api_key`, `face_id`, etc.) remains unchanged.
+  (PR [#4001](https://github.com/pipecat-ai/pipecat/pull/4001))
+
+- Update `pipecat-ai-small-webrtc-prebuilt` to `2.4.0`.
+  (PR [#4023](https://github.com/pipecat-ai/pipecat/pull/4023))
+
+- Nova Sonic assistant text transcripts are now delivered in real-time using
+  speculative text events instead of delayed final text events. Previously,
+  assistant text only arrived after all audio had finished playing, causing
+  laggy transcripts in client UIs. Speculative text arrives before each audio
+  chunk, providing text synchronized with what the bot is saying. This also
+  simplifies the internal text handling by removing the interruption re-push
+  hack and assistant text buffer.
+  (PR [#4042](https://github.com/pipecat-ai/pipecat/pull/4042))
+
+- Updated `daily-python` dependency to 0.25.0.
+  (PR [#4047](https://github.com/pipecat-ai/pipecat/pull/4047))
+
+- Added `enable_dialout` parameter to `configure()` in `pipecat.runner.daily`
+  to support dial-out rooms. Also narrowed misleading `Optional` type hints and
+  deduplicated token expiry calculation.
+  (PR [#4048](https://github.com/pipecat-ai/pipecat/pull/4048))
+
+- Extended `ProcessFrameResult` to stop strategies, allowing a stop strategy to
+  short-circuit evaluation of subsequent strategies by returning `STOP`.
+  (PR [#4064](https://github.com/pipecat-ai/pipecat/pull/4064))
+
+- `GradiumSTTService` now takes both an `encoding` and `sample_rate`
+  constructor argument which is assmebled in the class to form the
+  `input_format`. PCM accepts `8000`, `16000`, and `24000` Hz sample rates.
+  (PR [#4066](https://github.com/pipecat-ai/pipecat/pull/4066))
+
+- Improved `GradiumSTTService` transcription accuracy by reworking how text
+  fragments are accumulated and finalized. Previously, trailing words could be
+  dropped when the server's `flushed` response arrived before all text tokens
+  were delivered. The service now uses a short aggregation delay after flush to
+  capture trailing tokens, producing complete utterances.
+  (PR [#4066](https://github.com/pipecat-ai/pipecat/pull/4066))
+
+### Deprecated
+
+- `SimliVideoService.InputParams` is deprecated. Use the direct constructor
+  parameters `max_session_length`, `max_idle_time`, and `enable_logging`
+  instead.
+  (PR [#4001](https://github.com/pipecat-ai/pipecat/pull/4001))
+
+- Deprecated `LocalSmartTurnAnalyzerV2` and `LocalCoreMLSmartTurnAnalyzer`. Use
+  `LocalSmartTurnAnalyzerV3` instead. Instantiating these analyzers will now
+  emit a `DeprecationWarning`.
+  (PR [#4012](https://github.com/pipecat-ai/pipecat/pull/4012))
+
+- Deprecated `WakeCheckFilter` in favor of `WakePhraseUserTurnStartStrategy`.
+  (PR [#4064](https://github.com/pipecat-ai/pipecat/pull/4064))
+
+### Fixed
+
+- Fixed an issue where the default model for `OpenAILLMService` and
+  `AzureLLMService` was mistakenly reverted to `gpt-4o`. The defaults are now
+  restored to `gpt-4.1`.
+  (PR [#4000](https://github.com/pipecat-ai/pipecat/pull/4000))
+
+- Fixed a race condition where `EndTaskFrame` could cause the pipeline to shut
+  down before in-flight frames (e.g. LLM function call responses) finished
+  processing. `EndTaskFrame` and `StopTaskFrame` now flow through the pipeline
+  as `ControlFrame`s, ensuring all pending work is flushed before shutdown
+  begins. `CancelTaskFrame` and `InterruptionTaskFrame` remain immediate
+  (`SystemFrame`).
+  (PR [#4006](https://github.com/pipecat-ai/pipecat/pull/4006))
+
+- Fixed `ParallelPipeline` dropping or misordering frames during lifecycle
+  synchronization. Buffered frames are now flushed in the correct order
+  relative to synchronization frames (`StartFrame` goes first,
+  `EndFrame`/`CancelFrame` go after), and frames added to the buffer during
+  flush are also drained.
+  (PR [#4007](https://github.com/pipecat-ai/pipecat/pull/4007))
+
+- Fixed `TTSService` potentially canceling in-flight audio during shutdown. The
+  stop sequence now waits for all queued audio contexts to finish processing
+  before canceling the stop frame task.
+  (PR [#4007](https://github.com/pipecat-ai/pipecat/pull/4007))
+
+- Fixed `Language` enum values (e.g. `Language.ES`) not being converted to
+  service-specific codes when passed via
+  `settings=Service.Settings(language=Language.ES)` at init time. This caused
+  API errors (e.g. 400 from Rime) because the raw enum was sent instead of the
+  expected language code (e.g. `"spa"`). Runtime updates via
+  `UpdateSettingsFrame` were unaffected. The fix centralizes conversion in the
+  base `TTSService` and `STTService` classes so all services handle this
+  consistently.
+  (PR [#4024](https://github.com/pipecat-ai/pipecat/pull/4024))
+
+- Fixed `DeepgramSTTService` ignoring the `base_url` scheme when using `ws://`
+  or `http://`. Previously these were silently overwritten with `wss://` /
+  `https://`, breaking air-gapped or private deployments that don't use TLS.
+  All scheme choices (`wss://`, `https://`, `ws://`, `http://`, or bare
+  hostname) are now respected.
+  (PR [#4026](https://github.com/pipecat-ai/pipecat/pull/4026))
+
+- Fixed `LLMSwitcher.register_function()` and `register_direct_function()` not
+  accepting or forwarding the `timeout_secs` parameter.
+  (PR [#4037](https://github.com/pipecat-ai/pipecat/pull/4037))
+
+- Fixed empty user transcriptions in Nova Sonic causing spurious interruptions.
+  Previously, an empty transcription could trigger an interruption of the
+  assistant's response even though the user hadn't actually spoken.
+  (PR [#4042](https://github.com/pipecat-ai/pipecat/pull/4042))
+
+- Fixed `SonioxSTTService` and `OpenAIRealtimeSTTService` crash when language
+  parameters contain plain strings instead of `Language` enum values.
+  (PR [#4046](https://github.com/pipecat-ai/pipecat/pull/4046))
+
+- Fixed premature user turn stops caused by late transcriptions arriving
+  between turns. A stale transcript from the previous turn could persist into
+  the next turn and trigger a stop before the current turn's real transcript
+  arrived. Stop strategies are now reset at both turn start and turn stop to
+  prevent state from leaking across turn boundaries.
+  (PR [#4057](https://github.com/pipecat-ai/pipecat/pull/4057))
+
+- Fixed raw language strings like `"de-DE"` silently failing when passed to
+  TTS/STT services (e.g. ElevenLabs producing no audio). Raw strings now go
+  through the same `Language` enum resolution as enum values, so regional codes
+  like `"de-DE"` are properly converted to service-expected formats like
+  `"de"`. Unrecognized strings log a warning instead of failing silently.
+  (PR [#4058](https://github.com/pipecat-ai/pipecat/pull/4058))
+
+- Fixed Deepgram STT list-type settings (`keyterm`, `keywords`, `search`,
+  `redact`, `replace`) being stringified instead of passed as lists to the SDK,
+  which caused them to be sent as literal strings (e.g. `"['pipecat']"`) in the
+  WebSocket query params.
+  (PR [#4063](https://github.com/pipecat-ai/pipecat/pull/4063))
+
+- Fixed `MinWordsUserTurnStartStrategy` including text below the word threshold
+  in the output by resetting aggregation when the minimum word count is not
+  met.
+  (PR [#4064](https://github.com/pipecat-ai/pipecat/pull/4064))
+
+- Fixed audio overlap and potential dropped TTS content when multiple assistant
+  turns occur in quick succession. `TTSService` now flushes remaining text
+  before pausing frame processing on `LLMFullResponseEndFrame`/`EndFrame`,
+  instead of pausing first.
+  (PR [#4071](https://github.com/pipecat-ai/pipecat/pull/4071))
+
+### Security
+
+- Bumped PyJWT minimum version from 2.10.1 to 2.12.0 in the `livekit` extra to
+  address CVE-2026-32597 (GHSA-752w-5fwx-jx9f), where PyJWT <= 2.11.0 accepted
+  unknown `crit` header extensions.
+  (PR [#4035](https://github.com/pipecat-ai/pipecat/pull/4035))
+
 ## [0.0.105] - 2026-03-10
 
 ### Added

README.md

@@ -65,6 +65,10 @@ claude plugin marketplace add pipecat-ai/skills
 
 and install any of the available plugins.
 
+### 🧩 Community Integrations
+
+Build and share your own Pipecat service integrations! Browse existing [community integrations](https://docs.pipecat.ai/server/services/community-integrations) or check out our [guide](COMMUNITY_INTEGRATIONS.md) to create your own.
+
 ### 📺️ Pipecat TV Channel
 
 Catch new features, interviews, and how-tos on our [Pipecat TV](https://www.youtube.com/playlist?list=PLzU2zoMTQIHjqC3v4q2XVSR3hGSzwKFwH) channel.
@@ -94,6 +98,7 @@ Catch new features, interviews, and how-tos on our [Pipecat TV](https://www.yout
 | Vision & Image      | [fal](https://docs.pipecat.ai/server/services/image-generation/fal), [Google Imagen](https://docs.pipecat.ai/server/services/image-generation/google-imagen), [Moondream](https://docs.pipecat.ai/server/services/vision/moondream)                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        |
 | Audio Processing    | [Silero VAD](https://docs.pipecat.ai/server/utilities/audio/silero-vad-analyzer), [Krisp](https://docs.pipecat.ai/server/utilities/audio/krisp-filter), [Koala](https://docs.pipecat.ai/server/utilities/audio/koala-filter), [ai-coustics](https://docs.pipecat.ai/server/utilities/audio/aic-filter)                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                     |
 | Analytics & Metrics | [OpenTelemetry](https://docs.pipecat.ai/server/utilities/opentelemetry), [Sentry](https://docs.pipecat.ai/server/services/analytics/sentry)                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                |
+| Community           | [Browse community integrations →](https://docs.pipecat.ai/server/services/community-integrations)                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                             |
 
 📚 [View full services documentation →](https://docs.pipecat.ai/server/services/supported-services)
 

changelog/3449.changed.md

@@ -0,0 +1 @@
+- Renamed tracing span attributes to align with OpenTelemetry GenAI semantic conventions: `gen_ai.system` to `gen_ai.provider.name`, `system` to `gen_ai.system_instructions`, `gen_ai.usage.cache_read_input_tokens` to `gen_ai.usage.cache_read.input_tokens`, and `gen_ai.usage.cache_creation_input_tokens` to `gen_ai.usage.cache_creation.input_tokens`.

changelog/3449.fixed.md

@@ -0,0 +1 @@
+- Fixed stale `system_instruction` in LLM tracing spans by reading from `_settings.system_instruction` instead of the removed `_system_instruction` attribute.

changelog/3457.changed.md

@@ -1 +0,0 @@
-- Changed tool result JSON serialization to use `ensure_ascii=False`, preserving UTF-8 characters instead of escaping them. This reduces context size and token usage for non-English languages.

changelog/3991.changed.md

@@ -1 +0,0 @@
-- `OpenAIRealtimeSTTService`'s `noise_reduction` parameter is now part of `OpenAIRealtimeSTTSettings`, making it runtime-updatable via `STTUpdateSettingsFrame`. The direct `noise_reduction` init argument is deprecated as of 0.0.106.

changelog/3997.changed.md

@@ -1 +0,0 @@
-- Updated `sarvamai` dependency from `0.1.26a2` (alpha) to `0.1.26` (stable release).

changelog/4000.fixed.md

@@ -1 +0,0 @@
-- Fixed an issue where the default model for `OpenAILLMService` and `AzureLLMService` was mistakenly reverted to `gpt-4o`. The defaults are now restored to `gpt-4.1`.

changelog/4001.changed.md

@@ -1 +0,0 @@
-- `SimliVideoService` now extends `AIService` instead of `FrameProcessor`, aligning it with the HeyGen and Tavus video services. It supports `SimliVideoService.Settings(...)` for configuration and uses `start()`/`stop()`/`cancel()` lifecycle methods. Existing constructor usage (`api_key`, `face_id`, etc.) remains unchanged.

changelog/4001.deprecated.md

@@ -1 +0,0 @@
-- `SimliVideoService.InputParams` is deprecated. Use the direct constructor parameters `max_session_length`, `max_idle_time`, and `enable_logging` instead.

changelog/4004.added.md

@@ -1 +0,0 @@
-- Added optional `service` field to `ServiceUpdateSettingsFrame` (and its subclasses `LLMUpdateSettingsFrame`, `TTSUpdateSettingsFrame`, `STTUpdateSettingsFrame`) to target a specific service instance. When `service` is set, only the matching service applies the settings; others forward the frame unchanged. This enables updating a single service when multiple services of the same type exist in the pipeline.

changelog/4005.added.md

@@ -1 +0,0 @@
-- Added `sip_provider` and `room_geo` parameters to `configure()` in the Daily runner. These convenience parameters let callers specify a SIP provider name and geographic region directly without manually constructing `DailyRoomProperties` and `DailyRoomSipParams`.

changelog/4006.fixed.md

@@ -1 +0,0 @@
-- Fixed a race condition where `EndTaskFrame` could cause the pipeline to shut down before in-flight frames (e.g. LLM function call responses) finished processing. `EndTaskFrame` and `StopTaskFrame` now flow through the pipeline as `ControlFrame`s, ensuring all pending work is flushed before shutdown begins. `CancelTaskFrame` and `InterruptionTaskFrame` remain immediate (`SystemFrame`).

changelog/4007.fixed.2.md

@@ -1 +0,0 @@
-- Fixed `TTSService` potentially canceling in-flight audio during shutdown. The stop sequence now waits for all queued audio contexts to finish processing before canceling the stop frame task.

changelog/4007.fixed.md

@@ -1 +0,0 @@
-- Fixed `ParallelPipeline` dropping or misordering frames during lifecycle synchronization. Buffered frames are now flushed in the correct order relative to synchronization frames (`StartFrame` goes first, `EndFrame`/`CancelFrame` go after), and frames added to the buffer during flush are also drained.

changelog/4009.added.md

@@ -1 +0,0 @@
-- Added `PerplexityLLMAdapter` that automatically transforms conversation messages to satisfy Perplexity's stricter API constraints (strict role alternation, no non-initial system messages, last message must be user/tool). Previously, certain conversation histories could cause Perplexity API errors that didn't occur with OpenAI (`PerplexityLLMService` subclasses `OpenAILLMService` since Perplexity uses an OpenAI-compatible API).

changelog/4012.deprecated.md

@@ -1 +0,0 @@
-- Deprecated `LocalSmartTurnAnalyzerV2` and `LocalCoreMLSmartTurnAnalyzer`. Use `LocalSmartTurnAnalyzerV3` instead. Instantiating these analyzers will now emit a `DeprecationWarning`.

changelog/4023.changed.md

@@ -1 +0,0 @@
-- Update `pipecat-ai-small-webrtc-prebuilt` to `2.4.0`.

changelog/4024.fixed.md

@@ -1 +0,0 @@
-- Fixed `Language` enum values (e.g. `Language.ES`) not being converted to service-specific codes when passed via `settings=Service.Settings(language=Language.ES)` at init time. This caused API errors (e.g. 400 from Rime) because the raw enum was sent instead of the expected language code (e.g. `"spa"`). Runtime updates via `UpdateSettingsFrame` were unaffected. The fix centralizes conversion in the base `TTSService` and `STTService` classes so all services handle this consistently.

changelog/4029.added.2.md

@@ -0,0 +1 @@
+- Added `frame_order` parameter to `SyncParallelPipeline`. Set `frame_order=FrameOrder.PIPELINE` to push synchronized output frames in pipeline definition order (all frames from the first pipeline, then the second, etc.) instead of the default arrival order.

changelog/4029.added.md

@@ -0,0 +1 @@
+- Added `sync_with_audio` field to `OutputImageRawFrame`. When set to `True`, the output transport queues image frames with audio so they are displayed only after all preceding audio has been sent, enabling synchronized audio/image playback.

changelog/4029.fixed.3.md

@@ -0,0 +1 @@
+- Fixed `SyncParallelPipeline` breaking the Whisker debugger.

changelog/4029.fixed.md

@@ -0,0 +1 @@
+- Fixed `SyncParallelPipeline` race condition where concurrent SystemFrame processing (e.g. from RTVI) could corrupt sink queues and cause deadlocks. SystemFrames now take a fast path that passes them through without draining queued output.

changelog/4035.security.md

@@ -1 +0,0 @@
-- Bumped PyJWT minimum version from 2.10.1 to 2.12.0 in the `livekit` extra to address CVE-2026-32597 (GHSA-752w-5fwx-jx9f), where PyJWT <= 2.11.0 accepted unknown `crit` header extensions.

changelog/4037.fixed.md

@@ -1 +0,0 @@
-- Fixed `LLMSwitcher.register_function()` and `register_direct_function()` not accepting or forwarding the `timeout_secs` parameter.

changelog/4047.added.md

@@ -1 +0,0 @@
-- Added DTMF input event support to the Daily transport. Incoming DTMF tones are now received via Daily's `on_dtmf_event` callback and pushed into the pipeline as `InputDTMFFrame`, enabling bots to react to keypad presses from phone callers.

changelog/4047.changed.md

@@ -1 +0,0 @@
-- Updated `daily-python` dependency to 0.25.0.

changelog/4074.added.md

@@ -0,0 +1 @@
+- Added `OpenAIResponsesLLMService`, a new LLM service that uses the OpenAI Responses API. Supports streaming text, function calling, usage metrics, and out-of-band inference. Works with the universal `LLMContext` and `LLMContextAggregatorPair`. See `examples/foundational/07-interruptible-openai-responses.py` and `14-function-calling-openai-responses.py`.

changelog/4075.fixed.md

@@ -0,0 +1 @@
+- Fixed TTS frame ordering so that non-system frames always arrive in correct order relative to the `TTSStartedFrame`/`TTSAudioRawFrame`/`TTSStoppedFrame` sequence. Previously these frames could race ahead of or behind audio context frames, producing out-of-order output downstream.

changelog/4082.fixed.md

@@ -0,0 +1 @@
+- Fixed `SarvamTTSService` audio and error frames now route through `append_to_audio_context()` instead of `push_frame()`, ensuring correct behavior with audio contexts and interruptions.

changelog/4083.changed.md

@@ -0,0 +1 @@
+- `DeepgramSageMakerTTSService` now correctly routes audio through the base `TTSService` audio context queue. Audio frames are delivered via `append_to_audio_context()` instead of being pushed directly, enabling proper ordering, interruption handling, and start/stop frame lifecycle management. Interruptions now trigger a `Clear` message to Deepgram (flushing its text buffer) at the right time via `on_audio_context_interrupted`.

changelog/4090.fixed.md

@@ -0,0 +1 @@
+- Fixed audio frame ordering and interruption handling in Fish Audio, LMNT, Neuphonic, and Rime NonJson TTS services. These services were bypassing the base `TTSService` audio context serialization queue by pushing audio frames directly, which could cause out-of-order frames and broken interruptions during speech.

changelog/4093.fixed.md

@@ -0,0 +1,7 @@
+- Fixed Genesys AudioHook serializer to always include the `parameters` field in
+  protocol messages. The AudioHook protocol requires every message to carry a
+  `parameters` object (even if empty), but `_create_message` omitted it when no
+  parameters were provided. This caused clients that validate message structure
+  (including the Genesys reference implementation) to reject `pong` and
+  parameter-less `closed` responses, breaking server sequence tracking and
+  preventing `outputVariables` from reaching the Architect flow.

examples/foundational/05-sync-speech-and-image.py

@@ -16,11 +16,12 @@ from pipecat.frames.frames import (
     Frame,
     LLMContextFrame,
     LLMFullResponseStartFrame,
+    OutputImageRawFrame,
     TextFrame,
 )
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.sync_parallel_pipeline import SyncParallelPipeline
+from pipecat.pipeline.sync_parallel_pipeline import FrameOrder, SyncParallelPipeline
 from pipecat.pipeline.task import PipelineTask
 from pipecat.processors.aggregators.llm_context import LLMContext
 from pipecat.processors.aggregators.sentence import SentenceAggregator
@@ -30,6 +31,7 @@ from pipecat.runner.utils import create_transport
 from pipecat.services.cartesia.tts import CartesiaHttpTTSService
 from pipecat.services.fal.image import FalImageGenService
 from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.services.tts_service import TextAggregationMode
 from pipecat.transports.base_transport import BaseTransport, TransportParams
 from pipecat.transports.daily.transport import DailyParams
 
@@ -44,6 +46,18 @@ class MonthFrame(DataFrame):
         return f"{self.name}(month: {self.month})"
 
 
+class MarkImageForPlaybackSync(FrameProcessor):
+    """Marks output image frames to be synchronized with audio playback."""
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, OutputImageRawFrame):
+            frame.sync_with_audio = True
+
+        await self.push_frame(frame, direction)
+
+
 class MonthPrepender(FrameProcessor):
     def __init__(self):
         super().__init__()
@@ -101,6 +115,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
             settings=CartesiaHttpTTSService.Settings(
                 voice="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
             ),
+            # No need to aggregate by sentences (the default), as we already know we're getting full sentences
+            # (Otherwise the service will unnecessarily wait for follow-up input to confirm the sentence is complete,
+            #  which, sadly, actually breaks the synchronization mechanism)
+            text_aggregation_mode=TextAggregationMode.TOKEN,
         )
 
         imagegen = FalImageGenService(
@@ -119,17 +137,26 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         # that, each pipeline runs concurrently and `SyncParallelPipeline` will
         # wait for the input frame to be processed.
         #
+        # We use `FrameOrder.PIPELINE` so that each synchronized batch of output
+        # frames is pushed in the order the pipelines are listed: image first,
+        # then audio. This ensures the transport receives the image before the
+        # audio frames it should accompany.
+        #
         # Note that `SyncParallelPipeline` requires the last processor in each
         # of the pipelines to be synchronous. In this case, we use
-        # `CartesiaHttpTTSService` and `FalImageGenService` which make HTTP
+        # `FalImageGenService` and `CartesiaHttpTTSService` which make HTTP
         # requests and wait for the response.
         pipeline = Pipeline(
             [
                 llm,  # LLM
                 sentence_aggregator,  # Aggregates LLM output into full sentences
                 SyncParallelPipeline(  # Run pipelines in parallel aggregating the result
+                    [
+                        imagegen,  # Generate image
+                        MarkImageForPlaybackSync(),  # Mark image as needing sync w/audio during playback
+                    ],
                     [month_prepender, tts],  # Create "Month: sentence" and output audio
-                    [imagegen],  # Generate image
+                    frame_order=FrameOrder.PIPELINE,
                 ),
                 transport.output(),  # Transport output
             ]

examples/foundational/05a-local-sync-speech-and-image.py

@@ -1,202 +0,0 @@
-#
-# Copyright (c) 2024-2026, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import asyncio
-import os
-import sys
-import tkinter as tk
-
-import aiohttp
-from dotenv import load_dotenv
-from loguru import logger
-
-from pipecat.frames.frames import (
-    Frame,
-    LLMContextFrame,
-    OutputAudioRawFrame,
-    TextFrame,
-    TTSAudioRawFrame,
-    URLImageRawFrame,
-)
-from pipecat.pipeline.pipeline import Pipeline
-from pipecat.pipeline.runner import PipelineRunner
-from pipecat.pipeline.sync_parallel_pipeline import SyncParallelPipeline
-from pipecat.pipeline.task import PipelineTask
-from pipecat.processors.aggregators.llm_context import LLMContext
-from pipecat.processors.aggregators.sentence import SentenceAggregator
-from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
-from pipecat.services.cartesia.tts import CartesiaHttpTTSService
-from pipecat.services.fal.image import FalImageGenService
-from pipecat.services.openai.llm import OpenAILLMService
-from pipecat.transports.local.tk import TkLocalTransport, TkTransportParams
-
-load_dotenv(override=True)
-
-logger.remove(0)
-logger.add(sys.stderr, level="DEBUG")
-
-
-async def main():
-    async with aiohttp.ClientSession() as session:
-        tk_root = tk.Tk()
-        tk_root.title("Calendar")
-
-        runner = PipelineRunner()
-
-        async def get_month_data(month):
-            messages = [
-                {
-                    "role": "user",
-                    "content": f"Describe a nature photograph suitable for use in a calendar, for the month of {month}. Include only the image description with no preamble. Limit the description to one sentence, please.",
-                }
-            ]
-
-            class ImageDescription(FrameProcessor):
-                def __init__(self):
-                    super().__init__()
-                    self.text = ""
-
-                async def process_frame(self, frame: Frame, direction: FrameDirection):
-                    await super().process_frame(frame, direction)
-
-                    if isinstance(frame, TextFrame):
-                        self.text = frame.text
-                    await self.push_frame(frame, direction)
-
-            class AudioGrabber(FrameProcessor):
-                def __init__(self):
-                    super().__init__()
-                    self.audio = bytearray()
-                    self.frame = None
-
-                async def process_frame(self, frame: Frame, direction: FrameDirection):
-                    await super().process_frame(frame, direction)
-
-                    if isinstance(frame, TTSAudioRawFrame):
-                        self.audio.extend(frame.audio)
-                        self.frame = OutputAudioRawFrame(
-                            bytes(self.audio), frame.sample_rate, frame.num_channels
-                        )
-                    await self.push_frame(frame, direction)
-
-            class ImageGrabber(FrameProcessor):
-                def __init__(self):
-                    super().__init__()
-                    self.frame = None
-
-                async def process_frame(self, frame: Frame, direction: FrameDirection):
-                    await super().process_frame(frame, direction)
-
-                    if isinstance(frame, URLImageRawFrame):
-                        self.frame = frame
-                    await self.push_frame(frame, direction)
-
-            llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
-
-            tts = CartesiaHttpTTSService(
-                api_key=os.getenv("CARTESIA_API_KEY"),
-                settings=CartesiaHttpTTSService.Settings(
-                    voice="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
-                ),
-            )
-
-            imagegen = FalImageGenService(
-                settings=FalImageGenService.Settings(
-                    image_size="square_hd",
-                ),
-                aiohttp_session=session,
-                key=os.getenv("FAL_KEY"),
-            )
-
-            sentence_aggregator = SentenceAggregator()
-
-            description = ImageDescription()
-
-            audio_grabber = AudioGrabber()
-
-            image_grabber = ImageGrabber()
-
-            # With `SyncParallelPipeline` we synchronize audio and images by
-            # pushing them basically in order (e.g. I1 A1 A1 A1 I2 A2 A2 A2 A2
-            # I3 A3). To do that, each pipeline runs concurrently and
-            # `SyncParallelPipeline` will wait for the input frame to be
-            # processed.
-            #
-            # Note that `SyncParallelPipeline` requires the last processor in
-            # each of the pipelines to be synchronous. In this case, we use
-            # `CartesiaHttpTTSService` and `FalImageGenService` which make HTTP
-            # requests and wait for the response.
-            pipeline = Pipeline(
-                [
-                    llm,  # LLM
-                    sentence_aggregator,  # Aggregates LLM output into full sentences
-                    description,  # Store sentence
-                    SyncParallelPipeline(
-                        [tts, audio_grabber],  # Generate and store audio for the given sentence
-                        [imagegen, image_grabber],  # Generate and storeimage for the given sentence
-                    ),
-                ]
-            )
-
-            task = PipelineTask(pipeline)
-            await task.queue_frame(LLMContextFrame(LLMContext(messages)))
-            await task.stop_when_done()
-
-            await runner.run(task)
-
-            return {
-                "month": month,
-                "text": description.text,
-                "image": image_grabber.frame,
-                "audio": audio_grabber.frame,
-            }
-
-        transport = TkLocalTransport(
-            tk_root,
-            TkTransportParams(
-                audio_out_enabled=True,
-                video_out_enabled=True,
-                video_out_width=1024,
-                video_out_height=1024,
-            ),
-        )
-
-        pipeline = Pipeline([transport.output()])
-
-        task = PipelineTask(pipeline)
-
-        # We only specify a few months as we create tasks all at once and we
-        # might get rate limited otherwise.
-        months: list[str] = [
-            "January",
-            "February",
-        ]
-
-        # We create one task per month. This will be executed concurrently.
-        month_tasks = [asyncio.create_task(get_month_data(month)) for month in months]
-
-        # Now we wait for each month task in the order they're completed. The
-        # benefit is we'll have as little delay as possible before the first
-        # month, and likely no delay between months, but the months won't
-        # display in order.
-        async def show_images(month_tasks):
-            for month_data_task in asyncio.as_completed(month_tasks):
-                data = await month_data_task
-                await task.queue_frames([data["image"], data["audio"]])
-
-            await runner.stop_when_done()
-
-        async def run_tk():
-            while not task.has_finished():
-                tk_root.update()
-                tk_root.update_idletasks()
-                await asyncio.sleep(0.1)
-
-        await asyncio.gather(runner.run(task), show_images(month_tasks), run_tk())
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

examples/foundational/07-interruptible-openai-responses.py

@@ -0,0 +1,125 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import os
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import LLMRunFrame
+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.responses.llm import OpenAIResponsesLLMService
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.daily.transport import DailyParams
+from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+
+load_dotenv(override=True)
+
+# We use lambdas to defer transport parameter creation until the transport
+# type is selected at runtime.
+transport_params = {
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+    "twilio": lambda: FastAPIWebsocketParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+}
+
+
+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"),
+        settings=CartesiaTTSService.Settings(
+            voice="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
+        ),
+    )
+
+    llm = OpenAIResponsesLLMService(
+        api_key=os.getenv("OPENAI_API_KEY"),
+        settings=OpenAIResponsesLLMService.Settings(
+            system_instruction="You are a helpful assistant in a voice conversation. Your responses will be spoken aloud, so avoid emojis, bullet points, or other formatting that can't be spoken. Respond to what the user said in a creative, helpful, and brief way.",
+        ),
+    )
+
+    context = LLMContext()
+    user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
+        context,
+        user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
+    )
+
+    pipeline = Pipeline(
+        [
+            transport.input(),  # Transport user input
+            stt,
+            user_aggregator,  # User responses
+            llm,  # LLM
+            tts,  # TTS
+            transport.output(),  # Transport bot output
+            assistant_aggregator,  # 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")
+        # Kick off the conversation.
+        context.add_message(
+            {"role": "developer", "content": "Please introduce yourself to the user."}
+        )
+        await task.queue_frames([LLMRunFrame()])
+
+    @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()

examples/foundational/07e-interruptible-together.py

@@ -0,0 +1,125 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import os
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import LLMRunFrame
+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.together.llm import TogetherLLMService
+from pipecat.services.together.stt import TogetherSTTService
+from pipecat.services.together.tts import TogetherTTSService
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.daily.transport import DailyParams
+from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+
+load_dotenv(override=True)
+
+
+# We use lambdas to defer transport parameter creation until the transport
+# type is selected at runtime.
+transport_params = {
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+    "twilio": lambda: FastAPIWebsocketParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+}
+
+
+async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
+    logger.info(f"Starting bot")
+
+    stt = TogetherSTTService(api_key=os.getenv("TOGETHER_API_KEY"))
+
+    tts = TogetherTTSService(
+        api_key=os.getenv("TOGETHER_API_KEY"),
+        settings=TogetherTTSService.Settings(
+            voice="tara",
+        ),
+    )
+
+    llm = TogetherLLMService(
+        api_key=os.getenv("TOGETHER_API_KEY"),
+        settings=TogetherLLMService.Settings(
+            model="openai/gpt-oss-120b",
+            system_instruction="You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be spoken aloud, so avoid special characters that can't easily be spoken, such as emojis or bullet points. Respond to what the user said in a creative and helpful way.",
+        ),
+    )
+
+    context = LLMContext()
+    user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
+        context,
+        user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
+    )
+
+    pipeline = Pipeline(
+        [
+            transport.input(),  # Transport user input
+            stt,
+            user_aggregator,  # User responses
+            llm,  # LLM
+            tts,  # TTS
+            transport.output(),  # Transport bot output
+            assistant_aggregator,  # 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")
+        # Kick off the conversation.
+        context.add_message({"role": "user", "content": "Please introduce yourself"})
+        await task.queue_frames([LLMRunFrame()])
+
+    @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()

examples/foundational/10-wake-phrase.py

@@ -19,7 +19,6 @@ from pipecat.processors.aggregators.llm_response_universal import (
     LLMContextAggregatorPair,
     LLMUserAggregatorParams,
 )
-from pipecat.processors.filters.wake_check_filter import WakeCheckFilter
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
 from pipecat.services.cartesia.tts import CartesiaTTSService
@@ -28,6 +27,11 @@ 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_start import WakePhraseUserTurnStartStrategy
+from pipecat.turns.user_turn_strategies import (
+    UserTurnStrategies,
+    default_user_turn_start_strategies,
+)
 
 load_dotenv(override=True)
 
@@ -52,7 +56,12 @@ transport_params = {
 async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     logger.info(f"Starting bot")
 
-    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+    stt = DeepgramSTTService(
+        api_key=os.getenv("DEEPGRAM_API_KEY"),
+        settings=DeepgramSTTService.Settings(
+            keyterm=["pipecat"],
+        ),
+    )
 
     tts = CartesiaTTSService(
         api_key=os.getenv("CARTESIA_API_KEY"),
@@ -68,19 +77,28 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         ),
     )
 
-    hey_robot_filter = WakeCheckFilter(["hey robot", "hey, robot"])
-
     context = LLMContext()
     user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
         context,
-        user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
+        user_params=LLMUserAggregatorParams(
+            user_turn_strategies=UserTurnStrategies(
+                start=[
+                    WakePhraseUserTurnStartStrategy(
+                        phrases=["pipecat"],
+                        # Timeout before wake phrase must be spoken again
+                        timeout=5.0,
+                    ),
+                    *default_user_turn_start_strategies(),
+                ]
+            ),
+            vad_analyzer=SileroVADAnalyzer(),
+        ),
     )
 
     pipeline = Pipeline(
         [
             transport.input(),  # Transport user input
-            stt,  # STT
-            hey_robot_filter,  # Filter out speech not directed at the robot
+            stt,
             user_aggregator,  # User responses
             llm,  # LLM
             tts,  # TTS
@@ -102,12 +120,7 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     async def on_client_connected(transport, client):
         logger.info(f"Client connected")
         # Kick off the conversation.
-        context.add_message(
-            {
-                "role": "user",
-                "content": "Please introduce yourself. Tell the user they should say 'Hey Robot' before talking to you.",
-            }
-        )
+        context.add_message({"role": "user", "content": "Please introduce yourself to the user."})
         await task.queue_frames([LLMRunFrame()])
 
     @transport.event_handler("on_client_disconnected")

examples/foundational/12-describe-image-openai-responses.py

@@ -0,0 +1,139 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+
+import os
+
+from dotenv import load_dotenv
+from loguru import logger
+from PIL import Image
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import LLMRunFrame
+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.responses.llm import OpenAIResponsesLLMService
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.daily.transport import DailyParams
+
+load_dotenv(override=True)
+
+
+# We use lambdas to defer transport parameter creation until the transport
+# type is selected at runtime.
+transport_params = {
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+}
+
+
+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"),
+        settings=CartesiaTTSService.Settings(
+            voice="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
+        ),
+    )
+
+    llm = OpenAIResponsesLLMService(
+        api_key=os.getenv("OPENAI_API_KEY"),
+        settings=OpenAIResponsesLLMService.Settings(
+            system_instruction="You are a helpful assistant in a voice conversation. Your responses will be spoken aloud, so avoid emojis, bullet points, or other formatting that can't be spoken. Respond to what the user said in a creative, helpful, and brief way. You are also able to describe images.",
+        ),
+    )
+
+    context = LLMContext()
+    user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
+        context,
+        user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
+    )
+
+    pipeline = Pipeline(
+        [
+            transport.input(),  # Transport user input
+            stt,  # STT
+            user_aggregator,  # User responses
+            llm,  # LLM
+            tts,  # TTS
+            transport.output(),  # Transport bot output
+            assistant_aggregator,  # 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")
+
+        if not runner_args.body:
+            script_dir = os.path.dirname(__file__)
+            runner_args.body = {
+                "image_path": os.path.join(script_dir, "assets", "cat.jpg"),
+                "question": "Describe this image",
+            }
+
+        image_path = runner_args.body["image_path"]
+        question = runner_args.body["question"]
+
+        # Kick off the conversation.
+        image = Image.open(image_path)
+        message = await LLMContext.create_image_message(
+            image=image.tobytes(),
+            format="RGB",
+            size=image.size,
+            text=question,
+        )
+        context.add_message(message)
+        await task.queue_frames([LLMRunFrame()])
+
+    @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()

examples/foundational/13n-together-transcription.py

@@ -0,0 +1,83 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import os
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import Frame, TranscriptionFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineTask
+from pipecat.processors.audio.vad_processor import VADProcessor
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.runner.types import RunnerArguments
+from pipecat.runner.utils import create_transport
+from pipecat.services.together.stt import TogetherSTTService
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.daily.transport import DailyParams
+from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+
+load_dotenv(override=True)
+
+
+class TranscriptionLogger(FrameProcessor):
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, TranscriptionFrame):
+            print(f"Transcription: {frame.text}")
+
+        # Push all frames through
+        await self.push_frame(frame, direction)
+
+
+# We use lambdas to defer transport parameter creation until the transport
+# type is selected at runtime.
+transport_params = {
+    "daily": lambda: DailyParams(audio_in_enabled=True),
+    "twilio": lambda: FastAPIWebsocketParams(audio_in_enabled=True),
+    "webrtc": lambda: TransportParams(audio_in_enabled=True),
+}
+
+
+async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
+    logger.info(f"Starting bot")
+
+    stt = TogetherSTTService(api_key=os.getenv("TOGETHER_API_KEY"))
+
+    tl = TranscriptionLogger()
+    vad_processor = VADProcessor(vad_analyzer=SileroVADAnalyzer())
+
+    pipeline = Pipeline([transport.input(), vad_processor, stt, tl])
+
+    task = PipelineTask(
+        pipeline,
+        idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
+    )
+
+    @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()

examples/foundational/14-function-calling-openai-responses.py

@@ -0,0 +1,175 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import os
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.adapters.schemas.function_schema import FunctionSchema
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import LLMRunFrame, TTSSpeakFrame
+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.llm_service import FunctionCallParams
+from pipecat.services.openai.responses.llm import OpenAIResponsesLLMService
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.daily.transport import DailyParams
+from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+
+load_dotenv(override=True)
+
+
+async def fetch_weather_from_api(params: FunctionCallParams):
+    await params.result_callback({"conditions": "nice", "temperature": "75"})
+
+
+async def fetch_restaurant_recommendation(params: FunctionCallParams):
+    await params.result_callback({"name": "The Golden Dragon"})
+
+
+# We use lambdas to defer transport parameter creation until the transport
+# type is selected at runtime.
+transport_params = {
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+    "twilio": lambda: FastAPIWebsocketParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+}
+
+
+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"),
+        settings=CartesiaTTSService.Settings(
+            voice="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
+        ),
+    )
+
+    llm = OpenAIResponsesLLMService(
+        api_key=os.getenv("OPENAI_API_KEY"),
+        settings=OpenAIResponsesLLMService.Settings(
+            system_instruction="You are a helpful assistant in a voice conversation. Your responses will be spoken aloud, so avoid emojis, bullet points, or other formatting that can't be spoken. Respond to what the user said in a creative, helpful, and brief way.",
+        ),
+    )
+
+    # You can also register a function_name of None to get all functions
+    # sent to the same callback with an additional function_name parameter.
+    llm.register_function("get_current_weather", fetch_weather_from_api)
+    llm.register_function("get_restaurant_recommendation", fetch_restaurant_recommendation)
+
+    @llm.event_handler("on_function_calls_started")
+    async def on_function_calls_started(service, function_calls):
+        await tts.queue_frame(TTSSpeakFrame("Let me check on that."))
+
+    weather_function = FunctionSchema(
+        name="get_current_weather",
+        description="Get the current weather",
+        properties={
+            "location": {
+                "type": "string",
+                "description": "The city and state, e.g. San Francisco, CA",
+            },
+            "format": {
+                "type": "string",
+                "enum": ["celsius", "fahrenheit"],
+                "description": "The temperature unit to use. Infer this from the user's location.",
+            },
+        },
+        required=["location", "format"],
+    )
+    restaurant_function = FunctionSchema(
+        name="get_restaurant_recommendation",
+        description="Get a restaurant recommendation",
+        properties={
+            "location": {
+                "type": "string",
+                "description": "The city and state, e.g. San Francisco, CA",
+            },
+        },
+        required=["location"],
+    )
+    tools = ToolsSchema(standard_tools=[weather_function, restaurant_function])
+
+    context = LLMContext(tools=tools)
+    user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
+        context,
+        user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
+    )
+
+    pipeline = Pipeline(
+        [
+            transport.input(),
+            stt,
+            user_aggregator,
+            llm,
+            tts,
+            transport.output(),
+            assistant_aggregator,
+        ]
+    )
+
+    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")
+        # Kick off the conversation.
+        context.add_message(
+            {"role": "developer", "content": "Please introduce yourself to the user."}
+        )
+        await task.queue_frames([LLMRunFrame()])
+
+    @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()

examples/foundational/14d-function-calling-openai-responses-video.py

@@ -0,0 +1,195 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+
+import os
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.adapters.schemas.function_schema import FunctionSchema
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import LLMRunFrame, TTSSpeakFrame, UserImageRequestFrame
+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.processors.frame_processor import FrameDirection
+from pipecat.runner.types import RunnerArguments
+from pipecat.runner.utils import (
+    create_transport,
+    get_transport_client_id,
+    maybe_capture_participant_camera,
+)
+from pipecat.services.cartesia.tts import CartesiaTTSService
+from pipecat.services.deepgram.stt import DeepgramSTTService
+from pipecat.services.llm_service import FunctionCallParams
+from pipecat.services.openai.responses.llm import OpenAIResponsesLLMService
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.daily.transport import DailyParams
+
+load_dotenv(override=True)
+
+
+async def fetch_user_image(params: FunctionCallParams):
+    """Fetch the user image and push it to the LLM.
+
+    When called, this function pushes a UserImageRequestFrame upstream to the
+    transport. As a result, the transport will request the user image and push a
+    UserImageRawFrame downstream which will be added to the context by the LLM
+    assistant aggregator. The result_callback will be invoked once the image is
+    retrieved and processed.
+    """
+    user_id = params.arguments["user_id"]
+    question = params.arguments["question"]
+    logger.debug(f"Requesting image with user_id={user_id}, question={question}")
+
+    # Request a user image frame and indicate that it should be added to the
+    # context. Also associate it to the function call. Pass the result_callback
+    # so it can be invoked when the image is actually retrieved.
+    await params.llm.push_frame(
+        UserImageRequestFrame(
+            user_id=user_id,
+            text=question,
+            append_to_context=True,
+            function_name=params.function_name,
+            tool_call_id=params.tool_call_id,
+            result_callback=params.result_callback,
+        ),
+        FrameDirection.UPSTREAM,
+    )
+
+
+# We use lambdas to defer transport parameter creation until the transport
+# type is selected at runtime.
+transport_params = {
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        video_in_enabled=True,
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        video_in_enabled=True,
+    ),
+}
+
+
+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"),
+        settings=CartesiaTTSService.Settings(
+            voice="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
+        ),
+    )
+
+    llm = OpenAIResponsesLLMService(
+        api_key=os.getenv("OPENAI_API_KEY"),
+        settings=OpenAIResponsesLLMService.Settings(
+            system_instruction="You are a helpful assistant in a voice conversation. Your responses will be spoken aloud, so avoid emojis, bullet points, or other formatting that can't be spoken. Respond to what the user said in a creative, helpful, and brief way. You are able to describe images from the user camera.",
+        ),
+    )
+    llm.register_function("fetch_user_image", fetch_user_image)
+
+    @llm.event_handler("on_function_calls_started")
+    async def on_function_calls_started(service, function_calls):
+        await tts.queue_frame(TTSSpeakFrame("Let me check on that.", append_to_context=False))
+
+    fetch_image_function = FunctionSchema(
+        name="fetch_user_image",
+        description="Called when the user requests a description of their camera feed",
+        properties={
+            "user_id": {
+                "type": "string",
+                "description": "The ID of the user to grab the image from",
+            },
+            "question": {
+                "type": "string",
+                "description": "The question that the user is asking about the image",
+            },
+        },
+        required=["user_id", "question"],
+    )
+    tools = ToolsSchema(standard_tools=[fetch_image_function])
+
+    context = LLMContext(tools=tools)
+    user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
+        context,
+        user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
+    )
+
+    pipeline = Pipeline(
+        [
+            transport.input(),  # Transport user input
+            stt,  # STT
+            user_aggregator,  # User responses
+            llm,  # LLM
+            tts,  # TTS
+            transport.output(),  # Transport bot output
+            assistant_aggregator,  # 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")
+
+        await maybe_capture_participant_camera(transport, client)
+
+        client_id = get_transport_client_id(transport, client)
+
+        # Kick off the conversation.
+        context.add_message(
+            {
+                "role": "user",
+                "content": f"Please introduce yourself to the user. Use '{client_id}' as the user ID during function calls.",
+            }
+        )
+        await task.queue_frames([LLMRunFrame()])
+
+    @transport.event_handler("on_client_disconnected")
+    async def on_client_disconnected(transport, client):
+        logger.info(f"Client disconnected")
+        await task.cancel()
+
+    @tts.event_handler("on_tts_request")
+    async def on_tts_request(tts, context_id: str, text: str):
+        logger.debug(f"On TTS request: {context_id}: {text}")
+
+    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()

examples/foundational/20a-persistent-context-openai-responses.py

@@ -0,0 +1,249 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import glob
+import json
+import os
+from datetime import datetime
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.adapters.schemas.function_schema import FunctionSchema
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import LLMRunFrame, TTSSpeakFrame
+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.llm_service import FunctionCallParams
+from pipecat.services.openai.responses.llm import OpenAIResponsesLLMService
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.daily.transport import DailyParams
+from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+
+load_dotenv(override=True)
+
+
+BASE_FILENAME = "/tmp/pipecat_conversation_"
+
+
+async def fetch_weather_from_api(params: FunctionCallParams):
+    temperature = 75 if params.arguments["format"] == "fahrenheit" else 24
+    await params.result_callback(
+        {
+            "conditions": "nice",
+            "temperature": temperature,
+            "format": params.arguments["format"],
+            "timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"),
+        }
+    )
+
+
+async def get_saved_conversation_filenames(params: FunctionCallParams):
+    # Construct the full pattern including the BASE_FILENAME
+    full_pattern = f"{BASE_FILENAME}*.json"
+
+    # Use glob to find all matching files
+    matching_files = glob.glob(full_pattern)
+    logger.debug(f"matching files: {matching_files}")
+
+    await params.result_callback({"filenames": matching_files})
+
+
+async def save_conversation(params: FunctionCallParams):
+    timestamp = datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
+    filename = f"{BASE_FILENAME}{timestamp}.json"
+    logger.debug(
+        f"writing conversation to {filename}\n{json.dumps(params.context.get_messages(), indent=4)}"
+    )
+    try:
+        with open(filename, "w") as file:
+            messages = params.context.get_messages()
+            # remove the last message, which is the instruction we just gave to save the conversation
+            messages.pop()
+            json.dump(messages, file, indent=2)
+        await params.result_callback({"success": True})
+    except Exception as e:
+        await params.result_callback({"success": False, "error": str(e)})
+
+
+async def load_conversation(params: FunctionCallParams):
+    global tts
+    filename = params.arguments["filename"]
+    logger.debug(f"loading conversation from {filename}")
+    try:
+        with open(filename, "r") as file:
+            params.context.set_messages(json.load(file))
+            logger.debug(
+                f"loaded conversation from {filename}\n{json.dumps(params.context.get_messages(), indent=4)}"
+            )
+        await params.llm.queue_frame(TTSSpeakFrame("Ok, I've loaded that conversation."))
+    except Exception as e:
+        await params.result_callback({"success": False, "error": str(e)})
+
+
+system_instruction = "You are a helpful assistant in a voice conversation. Your responses will be spoken aloud, so avoid emojis, bullet points, or other formatting that can't be spoken. Respond to what the user said in a creative, helpful, and brief way."
+
+weather_function = FunctionSchema(
+    name="get_current_weather",
+    description="Get the current weather",
+    properties={
+        "location": {
+            "type": "string",
+            "description": "The city and state, e.g. San Francisco, CA",
+        },
+        "format": {
+            "type": "string",
+            "enum": ["celsius", "fahrenheit"],
+            "description": "The temperature unit to use. Infer this from the users location.",
+        },
+    },
+    required=["location", "format"],
+)
+
+save_conversation_function = FunctionSchema(
+    name="save_conversation",
+    description="Save the current conversation. Use this function to persist the current conversation to external storage.",
+    properties={},
+    required=[],
+)
+
+get_filenames_function = FunctionSchema(
+    name="get_saved_conversation_filenames",
+    description="Get a list of saved conversation histories. Returns a list of filenames. Each filename includes a date and timestamp. Each file is conversation history that can be loaded into this session.",
+    properties={},
+    required=[],
+)
+
+load_conversation_function = FunctionSchema(
+    name="load_conversation",
+    description="Load a conversation history. Use this function to load a conversation history into the current session.",
+    properties={
+        "filename": {
+            "type": "string",
+            "description": "The filename of the conversation history to load.",
+        }
+    },
+    required=["filename"],
+)
+
+tools = ToolsSchema(
+    standard_tools=[
+        weather_function,
+        save_conversation_function,
+        get_filenames_function,
+        load_conversation_function,
+    ]
+)
+
+
+# We use lambdas to defer transport parameter creation until the transport
+# type is selected at runtime.
+transport_params = {
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+    "twilio": lambda: FastAPIWebsocketParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+}
+
+
+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"),
+        settings=CartesiaTTSService.Settings(
+            voice="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
+        ),
+    )
+
+    llm = OpenAIResponsesLLMService(
+        api_key=os.getenv("OPENAI_API_KEY"),
+        settings=OpenAIResponsesLLMService.Settings(
+            system_instruction=system_instruction,
+        ),
+    )
+
+    # you can either register a single function for all function calls, or specific functions
+    # llm.register_function(None, fetch_weather_from_api)
+    llm.register_function("get_current_weather", fetch_weather_from_api)
+    llm.register_function("save_conversation", save_conversation)
+    llm.register_function("get_saved_conversation_filenames", get_saved_conversation_filenames)
+    llm.register_function("load_conversation", load_conversation)
+
+    context = LLMContext(tools=tools)
+    user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
+        context,
+        user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
+    )
+
+    pipeline = Pipeline(
+        [
+            transport.input(),  # Transport user input
+            stt,  # STT
+            user_aggregator,
+            llm,  # LLM
+            tts,
+            transport.output(),  # Transport bot output
+            assistant_aggregator,
+        ]
+    )
+
+    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")
+        # Kick off the conversation.
+        await task.queue_frames([LLMRunFrame()])
+
+    @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()

examples/foundational/20a-persistent-context-openai.py

@@ -116,7 +116,7 @@ weather_function = FunctionSchema(
 
 save_conversation_function = FunctionSchema(
     name="save_conversation",
-    description="Save the current conversatione. Use this function to persist the current conversation to external storage.",
+    description="Save the current conversation. Use this function to persist the current conversation to external storage.",
     properties={},
     required=[],
 )

examples/foundational/20b-persistent-context-openai-realtime-beta.py

@@ -119,7 +119,7 @@ tools = [
     {
         "type": "function",
         "name": "save_conversation",
-        "description": "Save the current conversatione. Use this function to persist the current conversation to external storage.",
+        "description": "Save the current conversation. Use this function to persist the current conversation to external storage.",
         "parameters": {
             "type": "object",
             "properties": {},

examples/foundational/20b-persistent-context-openai-realtime.py

@@ -125,7 +125,7 @@ tools = ToolsSchema(
         ),
         FunctionSchema(
             name="save_conversation",
-            description="Save the current conversatione. Use this function to persist the current conversation to external storage.",
+            description="Save the current conversation. Use this function to persist the current conversation to external storage.",
             properties={},
             required=[],
         ),

examples/foundational/55zi-update-settings-openai-responses-llm.py

@@ -0,0 +1,127 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import LLMRunFrame, LLMUpdateSettingsFrame
+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.responses.llm import OpenAIResponsesLLMService
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.daily.transport import DailyParams
+from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams
+
+load_dotenv(override=True)
+
+transport_params = {
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+    "twilio": lambda: FastAPIWebsocketParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+}
+
+
+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"),
+        settings=CartesiaTTSService.Settings(
+            voice="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
+        ),
+    )
+
+    llm = OpenAIResponsesLLMService(
+        api_key=os.getenv("OPENAI_API_KEY"),
+        settings=OpenAIResponsesLLMService.Settings(
+            system_instruction="You are a helpful assistant in a voice conversation. Your responses will be spoken aloud, so avoid emojis, bullet points, or other formatting that can't be spoken. Respond to what the user said in a creative, helpful, and brief way.",
+        ),
+    )
+
+    context = LLMContext()
+    user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
+        context,
+        user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
+    )
+
+    pipeline = Pipeline(
+        [
+            transport.input(),
+            stt,
+            user_aggregator,
+            llm,
+            tts,
+            transport.output(),
+            assistant_aggregator,
+        ]
+    )
+
+    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")
+        context.add_message({"role": "user", "content": "Please introduce yourself to the user."})
+        await task.queue_frames([LLMRunFrame()])
+
+        await asyncio.sleep(10)
+        logger.info("Updating OpenAI LLM settings: temperature=0.1")
+        await task.queue_frame(
+            LLMUpdateSettingsFrame(delta=OpenAIResponsesLLMService.Settings(temperature=0.1))
+        )
+
+    @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()

pyproject.toml

@@ -118,7 +118,7 @@ soundfile = [ "soundfile~=0.13.1" ]
 speechmatics = [ "speechmatics-voice[smart]~=0.2.8" ]
 strands = [ "strands-agents>=1.9.1,<2" ]
 tavus=[]
-together = []
+together = [ "pipecat-ai[websockets-base]" ]
 tracing = [ "opentelemetry-sdk>=1.33.0,<2", "opentelemetry-api>=1.33.0,<2", "opentelemetry-instrumentation>=0.54b0,<1" ]
 ultravox = [ "pipecat-ai[websockets-base]" ]
 webrtc = [ "aiortc>=1.14.0,<2", "opencv-python>=4.11.0.86,<5" ]

scripts/evals/run-release-evals.py

@@ -147,12 +147,14 @@ TESTS_07 = [
     ("07zi-interruptible-piper.py", EVAL_SIMPLE_MATH),
     ("07zj-interruptible-kokoro.py", EVAL_SIMPLE_MATH),
     ("07zk-interruptible-resembleai.py", EVAL_SIMPLE_MATH),
+    ("07-interruptible-openai-responses.py", EVAL_SIMPLE_MATH),
     # Needs a local XTTS docker instance running.
     # ("07i-interruptible-xtts.py", EVAL_SIMPLE_MATH),
 ]
 
 TESTS_12 = [
     ("12-describe-image-openai.py", EVAL_VISION_IMAGE(eval_speaks_first=True)),
+    ("12-describe-image-openai-responses.py", EVAL_VISION_IMAGE(eval_speaks_first=True)),
     ("12a-describe-image-anthropic.py", EVAL_VISION_IMAGE(eval_speaks_first=True)),
     ("12b-describe-image-aws.py", EVAL_VISION_IMAGE(eval_speaks_first=True)),
     ("12c-describe-image-gemini-flash.py", EVAL_VISION_IMAGE(eval_speaks_first=True)),
@@ -184,12 +186,15 @@ TESTS_14 = [
     ("14v-function-calling-openai.py", EVAL_WEATHER),
     ("14w-function-calling-mistral.py", EVAL_WEATHER),
     ("14x-function-calling-openpipe.py", EVAL_WEATHER),
+    ("14-function-calling-openai-responses.py", EVAL_WEATHER),
+    ("14-function-calling-openai-responses.py", EVAL_WEATHER_AND_RESTAURANT),
     # Video
     ("14d-function-calling-anthropic-video.py", EVAL_VISION_CAMERA),
     ("14d-function-calling-aws-video.py", EVAL_VISION_CAMERA),
     ("14d-function-calling-gemini-flash-video.py", EVAL_VISION_CAMERA),
     ("14d-function-calling-moondream-video.py", EVAL_VISION_CAMERA),
     ("14d-function-calling-openai-video.py", EVAL_VISION_CAMERA),
+    ("14d-function-calling-openai-responses-video.py", EVAL_VISION_CAMERA),
     # Currently not working.
     # ("14c-function-calling-together.py", EVAL_WEATHER),
     # ("14l-function-calling-deepseek.py", EVAL_WEATHER),

src/pipecat/adapters/services/open_ai_responses_adapter.py

@@ -0,0 +1,254 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""OpenAI Responses API adapter for Pipecat."""
+
+import copy
+from typing import Any, Dict, List, Optional, TypedDict
+
+from loguru import logger
+from openai._types import NotGiven as OpenAINotGiven
+from openai.types.responses import FunctionToolParam, ResponseInputItemParam
+
+from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.processors.aggregators.llm_context import (
+    LLMContext,
+    LLMContextMessage,
+    LLMSpecificMessage,
+    NotGiven,
+)
+
+
+class OpenAIResponsesLLMInvocationParams(TypedDict, total=False):
+    """Context-based parameters for invoking OpenAI Responses API."""
+
+    input: List[ResponseInputItemParam]
+    tools: List[FunctionToolParam] | OpenAINotGiven
+    instructions: str
+
+
+class OpenAIResponsesLLMAdapter(BaseLLMAdapter[OpenAIResponsesLLMInvocationParams]):
+    """OpenAI Responses API adapter for Pipecat.
+
+    Handles:
+
+    - Converting LLMContext messages to Responses API input items
+    - Converting Pipecat's standardized tools schema to Responses API function tool format
+    - Extracting and sanitizing messages from the LLM context for logging
+    """
+
+    def __init__(self):
+        """Initialize the adapter."""
+        super().__init__()
+        self._warned_system_instruction = False
+
+    @property
+    def id_for_llm_specific_messages(self) -> str:
+        """Get the identifier used in LLMSpecificMessage instances."""
+        return "openai_responses"
+
+    def get_llm_invocation_params(
+        self,
+        context: LLMContext,
+        *,
+        system_instruction: Optional[str] = None,
+    ) -> OpenAIResponsesLLMInvocationParams:
+        """Get Responses API invocation parameters from a universal LLM context.
+
+        Args:
+            context: The LLM context containing messages, tools, etc.
+            system_instruction: Optional system instruction from service settings.
+
+        Returns:
+            Dictionary of parameters for the Responses API.
+        """
+        messages = self.get_messages(context)
+        input_items = self._convert_messages_to_input(messages)
+
+        params: OpenAIResponsesLLMInvocationParams = {
+            "input": input_items,
+            "tools": self.from_standard_tools(context.tools),
+        }
+
+        if system_instruction:
+            # Compatibility: The Responses API requires at least one input
+            # message when instructions are provided. Contexts that worked with
+            # OpenAILLMService (system_instruction + empty messages) need the
+            # instructions converted to an initial developer message.
+            #
+            # NOTE: if/when we support `previous_response_id` and/or
+            # `conversation_id`, we'll need to revisit this logic, as it'll
+            # be legit to provide instructions without input items. Worth
+            # noting that OpenAI's docs suggest these parameters are primarily
+            # for development convenience rather than performance (the model
+            # still processes the full context), and come with the tradeoff
+            # of requiring OpenAI-side 30-day conversation storage, which may
+            # not be desirable for many users. But it could give folks an easy
+            # way to store/switch between conversations without needing to
+            # manage that storage themselves.
+            if not input_items:
+                params["input"] = [{"role": "developer", "content": system_instruction}]
+            else:
+                params["instructions"] = system_instruction
+
+        return params
+
+    def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[FunctionToolParam]:
+        """Convert function schemas to Responses API function tool format.
+
+        Args:
+            tools_schema: The Pipecat tools schema to convert.
+
+        Returns:
+            List of Responses API function tool definitions.
+        """
+        functions_schema = tools_schema.standard_tools
+        result = []
+        for func in functions_schema:
+            d = func.to_default_dict()
+            tool: FunctionToolParam = {
+                "type": "function",
+                "name": d["name"],
+                "parameters": d.get("parameters", {}),
+                "strict": d.get("strict", None),
+            }
+            if "description" in d:
+                tool["description"] = d["description"]
+            result.append(tool)
+        return result
+
+    def get_messages_for_logging(self, context: LLMContext) -> List[Dict[str, Any]]:
+        """Get messages from context in a format ready for logging.
+
+        Removes or truncates sensitive data like image content for safe logging.
+
+        Args:
+            context: The LLM context containing messages.
+
+        Returns:
+            List of messages in a format ready for logging.
+        """
+        msgs = []
+        for message in self.get_messages(context):
+            msg = copy.deepcopy(message)
+            if "content" in msg:
+                if isinstance(msg["content"], list):
+                    for item in msg["content"]:
+                        if item.get("type") == "image_url":
+                            if item["image_url"]["url"].startswith("data:image/"):
+                                item["image_url"]["url"] = "data:image/..."
+                        if item.get("type") == "input_audio":
+                            item["input_audio"]["data"] = "..."
+            msgs.append(msg)
+        return msgs
+
+    def _convert_messages_to_input(
+        self, messages: List[LLMContextMessage]
+    ) -> List[ResponseInputItemParam]:
+        """Convert LLMContext messages to Responses API input items.
+
+        Args:
+            messages: Messages from the LLMContext.
+
+        Returns:
+            List of Responses API input items.
+        """
+        result: List[ResponseInputItemParam] = []
+        is_first = True
+
+        for message in messages:
+            if isinstance(message, LLMSpecificMessage):
+                result.append(message.message)
+                is_first = False
+                continue
+
+            role = message.get("role")
+
+            if role == "system":
+                if is_first and not self._warned_system_instruction:
+                    logger.warning(
+                        "System messages in LLMContext are converted to 'developer' role for the "
+                        "Responses API. Consider using settings.system_instruction instead, which "
+                        "maps to the 'instructions' parameter."
+                    )
+                    self._warned_system_instruction = True
+                content = message.get("content", "")
+                if isinstance(content, list):
+                    content = self._convert_multimodal_content(content)
+                result.append({"role": "developer", "content": content})
+
+            elif role == "user":
+                content = message.get("content", "")
+                if isinstance(content, list):
+                    content = self._convert_multimodal_content(content)
+                result.append({"role": "user", "content": content})
+
+            elif role == "assistant":
+                tool_calls = message.get("tool_calls")
+                if tool_calls:
+                    for tc in tool_calls:
+                        func = tc.get("function", {})
+                        result.append(
+                            {
+                                "type": "function_call",
+                                "call_id": tc.get("id", ""),
+                                "name": func.get("name", ""),
+                                "arguments": func.get("arguments", ""),
+                            }
+                        )
+                else:
+                    content = message.get("content", "")
+                    if isinstance(content, list):
+                        content = self._convert_multimodal_content(content)
+                    result.append({"role": "assistant", "content": content})
+
+            elif role == "tool":
+                content = message.get("content", "")
+                if not isinstance(content, str):
+                    content = str(content)
+                result.append(
+                    {
+                        "type": "function_call_output",
+                        "call_id": message.get("tool_call_id", ""),
+                        "output": content,
+                    }
+                )
+
+            is_first = False
+
+        return result
+
+    def _convert_multimodal_content(self, content: list) -> list:
+        """Convert multimodal content parts to Responses API format.
+
+        Args:
+            content: List of content parts from the LLMContext message.
+
+        Returns:
+            List of content parts in Responses API format.
+        """
+        result = []
+        for part in content:
+            part_type = part.get("type")
+            if part_type == "text":
+                result.append({"type": "input_text", "text": part.get("text", "")})
+            elif part_type == "image_url":
+                image_url_obj = part.get("image_url", {})
+                result.append(
+                    {
+                        "type": "input_image",
+                        "image_url": image_url_obj.get("url", ""),
+                        "detail": image_url_obj.get("detail", "auto"),
+                    }
+                )
+            else:
+                # Pass through other types as-is. Note: "input_audio" is not
+                # yet supported by the Responses API (coming soon per OpenAI
+                # docs) but the LLMContext format already matches the expected
+                # shape, so it should work once support is enabled.
+                result.append(part)
+        return result

src/pipecat/frames/frames.py

@@ -274,8 +274,16 @@ class OutputImageRawFrame(DataFrame, ImageRawFrame):
     An image that will be shown by the transport. If the transport supports
     multiple video destinations (e.g. multiple video tracks) the destination
     name can be specified in transport_destination.
+
+    Parameters:
+        sync_with_audio: If True, the image is queued with audio frames so
+            it is only displayed after all preceding audio has been sent.
+            Defaults to False (image is displayed immediately when the output
+            transport receives it).
     """
 
+    sync_with_audio: bool = field(default=False, init=False)
+
     def __str__(self):
         pts = format_pts(self.pts)
         return f"{self.name}(pts: {pts}, destination: {self.transport_destination}, size: {self.size}, format: {self.format})"

src/pipecat/pipeline/sync_parallel_pipeline.py

@@ -4,15 +4,21 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-"""Synchronous parallel pipeline implementation for concurrent frame processing.
+"""Synchronized parallel pipeline that holds output until all branches finish.
 
-This module provides a pipeline that processes frames through multiple parallel
-pipelines simultaneously, synchronizing their output to maintain frame ordering
-and prevent duplicate processing.
+A SyncParallelPipeline fans each inbound frame out to multiple parallel pipelines
+and waits for every pipeline to finish processing before releasing any of the
+resulting output frames. This ensures that all frames produced in response to a
+single input frame are emitted together.
+
+System frames (except EndFrame) are exempt from this synchronization — they pass
+straight through without waiting, since they are expected to race ahead of
+regular data frames.
 """
 
 import asyncio
 from dataclasses import dataclass
+from enum import Enum
 from itertools import chain
 from typing import List
 
@@ -24,22 +30,42 @@ from pipecat.pipeline.pipeline import Pipeline
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup
 
 
+class FrameOrder(Enum):
+    """Controls the order in which synchronized frames are pushed downstream.
+
+    When multiple parallel pipelines produce output for the same input frame,
+    this setting determines the order in which those output frames are pushed.
+
+    Attributes:
+        ARRIVAL: Frames are pushed in the order they arrive from any pipeline.
+            This is the default and matches the behavior of prior versions.
+        PIPELINE: Frames are pushed in pipeline definition order — all frames
+            from the first pipeline are pushed, then all frames from the second
+            pipeline, and so on. Useful when the relative ordering between
+            pipelines matters (e.g. ensuring image frames precede audio frames).
+    """
+
+    ARRIVAL = "arrival"
+    PIPELINE = "pipeline"
+
+
 @dataclass
 class SyncFrame(ControlFrame):
-    """Control frame used to synchronize parallel pipeline processing.
+    """Sentinel frame used to detect when a parallel pipeline has finished processing.
 
-    This frame is sent through parallel pipelines to determine when the
-    internal pipelines have finished processing a batch of frames.
+    After sending a real frame into a parallel pipeline, a SyncFrame is sent
+    behind it. When the SyncFrame emerges from the pipeline's output, we know
+    all output frames for the preceding input have been produced.
     """
 
     pass
 
 
 class SyncParallelPipelineSource(FrameProcessor):
-    """Source processor for synchronous parallel pipeline processing.
+    """Bookend processor placed at the start of each parallel pipeline.
 
-    Routes frames to parallel pipelines and collects upstream responses
-    for synchronization purposes.
+    Forwards downstream frames into the pipeline and captures upstream frames
+    into a queue so the parent SyncParallelPipeline can release them later.
     """
 
     def __init__(self, upstream_queue: asyncio.Queue):
@@ -68,10 +94,11 @@ class SyncParallelPipelineSource(FrameProcessor):
 
 
 class SyncParallelPipelineSink(FrameProcessor):
-    """Sink processor for synchronous parallel pipeline processing.
+    """Bookend processor placed at the end of each parallel pipeline.
 
-    Collects downstream frames from parallel pipelines and routes
-    upstream frames back through the pipeline.
+    Captures downstream output frames into a queue so the parent
+    SyncParallelPipeline can release them later, and forwards upstream
+    frames back through the pipeline.
     """
 
     def __init__(self, downstream_queue: asyncio.Queue):
@@ -100,29 +127,44 @@ class SyncParallelPipelineSink(FrameProcessor):
 
 
 class SyncParallelPipeline(BasePipeline):
-    """Pipeline that processes frames through multiple parallel pipelines synchronously.
+    """Fans each input frame to parallel pipelines then holds output until every pipeline finishes.
 
-    Creates multiple parallel processing paths that all receive the same input frames
-    and produces synchronized output. Each parallel path is a separate pipeline that
-    processes frames independently, with synchronization points to ensure consistent
-    ordering and prevent duplicate frame processing.
+    For each inbound frame the pipeline:
 
-    The pipeline uses SyncFrame control frames to coordinate between parallel paths
-    and ensure all paths have completed processing before moving to the next frame.
+    1. Sends the frame into every parallel pipeline.
+    2. Sends a ``SyncFrame`` sentinel behind it in each pipeline.
+    3. Waits until every pipeline has produced its ``SyncFrame``, meaning all
+       output for that input is ready.
+    4. Releases the collected output frames (deduplicating by frame id, since
+       the same frame may emerge from more than one branch).
+
+    System frames (except ``EndFrame``) bypass this mechanism entirely — they are
+    forwarded through each pipeline and pushed immediately, since system frames
+    are expected to race ahead of regular data frames.
+
+    By default, output frames are pushed in the order they arrive from any pipeline
+    (``FrameOrder.ARRIVAL``). Set ``frame_order=FrameOrder.PIPELINE`` to push frames
+    in pipeline definition order instead — all output from the first pipeline, then
+    the second, and so on.
     """
 
-    def __init__(self, *args):
+    def __init__(self, *args, frame_order: FrameOrder = FrameOrder.ARRIVAL):
         """Initialize the synchronous parallel pipeline.
 
         Args:
-            *args: Variable number of processor lists, each representing a parallel pipeline path.
-                   Each argument should be a list of FrameProcessor instances.
+            *args: Variable number of processor lists, each representing a parallel
+                pipeline path. Each argument should be a list of FrameProcessor instances.
+            frame_order: Controls the order in which synchronized output frames are
+                pushed. ``FrameOrder.ARRIVAL`` (default) pushes frames in the order they arrive.
+                ``FrameOrder.PIPELINE`` pushes all frames from the first pipeline
+                before the second, and so on.
 
         Raises:
             Exception: If no arguments are provided.
             TypeError: If any argument is not a list of processors.
         """
         super().__init__()
+        self._frame_order = frame_order
 
         if len(args) == 0:
             raise Exception(f"SyncParallelPipeline needs at least one argument")
@@ -184,7 +226,7 @@ class SyncParallelPipeline(BasePipeline):
         Returns:
             The list of entry processors.
         """
-        return self._sources
+        return [s["processor"] for s in self._sources]
 
     def processors_with_metrics(self) -> List[FrameProcessor]:
         """Collect processors that can generate metrics from all parallel pipelines.
@@ -209,11 +251,11 @@ class SyncParallelPipeline(BasePipeline):
         await asyncio.gather(*[p.cleanup() for p in self._pipelines])
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
-        """Process frames through all parallel pipelines with synchronization.
+        """Send a frame through all parallel pipelines and release output once all finish.
 
-        Distributes frames to all parallel pipelines and synchronizes their output
-        to maintain proper ordering and prevent duplicate processing. Uses SyncFrame
-        control frames to coordinate between parallel paths.
+        System frames (except EndFrame) skip synchronization and pass straight
+        through. All other frames are fanned out to every pipeline, and output is
+        held until every pipeline signals completion (via SyncFrame).
 
         Args:
             frame: The frame to process.
@@ -221,60 +263,102 @@ class SyncParallelPipeline(BasePipeline):
         """
         await super().process_frame(frame, direction)
 
+        # SystemFrames (but not EndFrame) are simply passed through all
+        # internal pipelines without draining queued output. This avoids
+        # the race condition where a SystemFrame's wait_for_sync steals
+        # frames from a concurrent non-SystemFrame's wait_for_sync.
+        if isinstance(frame, SystemFrame) and not isinstance(frame, EndFrame):
+            if direction == FrameDirection.UPSTREAM:
+                for s in self._sinks:
+                    await s["processor"].process_frame(frame, direction)
+            elif direction == FrameDirection.DOWNSTREAM:
+                for s in self._sources:
+                    await s["processor"].process_frame(frame, direction)
+            await self.push_frame(frame, direction)
+            return
+
+        use_pipeline_order = self._frame_order == FrameOrder.PIPELINE
+
         # The last processor of each pipeline needs to be synchronous otherwise
-        # this element won't work. Since, we know it should be synchronous we
+        # this element won't work. Since we know it should be synchronous we
         # push a SyncFrame. Since frames are ordered we know this frame will be
         # pushed after the synchronous processor has pushed its data allowing us
-        # to synchrnonize all the internal pipelines by waiting for the
+        # to synchronize all the internal pipelines by waiting for the
         # SyncFrame in all of them.
+        #
+        # In ARRIVAL mode, output frames are put onto a shared main_queue as
+        # they arrive. In PIPELINE mode, they are accumulated in a per-pipeline
+        # list and returned so the caller can drain them in definition order.
         async def wait_for_sync(
             obj, main_queue: asyncio.Queue, frame: Frame, direction: FrameDirection
-        ):
+        ) -> list[Frame]:
             processor = obj["processor"]
             queue = obj["queue"]
+            output_frames: list[Frame] = []
 
             await processor.process_frame(frame, direction)
 
-            if isinstance(frame, (SystemFrame, EndFrame)):
+            if isinstance(frame, EndFrame):
                 new_frame = await queue.get()
-                if isinstance(new_frame, (SystemFrame, EndFrame)):
-                    await main_queue.put(new_frame)
-                else:
-                    while not isinstance(new_frame, (SystemFrame, EndFrame)):
+                if isinstance(new_frame, EndFrame):
+                    if use_pipeline_order:
+                        output_frames.append(new_frame)
+                    else:
                         await main_queue.put(new_frame)
+                else:
+                    while not isinstance(new_frame, EndFrame):
+                        if use_pipeline_order:
+                            output_frames.append(new_frame)
+                        else:
+                            await main_queue.put(new_frame)
                         queue.task_done()
                         new_frame = await queue.get()
             else:
                 await processor.process_frame(SyncFrame(), direction)
                 new_frame = await queue.get()
                 while not isinstance(new_frame, SyncFrame):
-                    await main_queue.put(new_frame)
+                    if use_pipeline_order:
+                        output_frames.append(new_frame)
+                    else:
+                        await main_queue.put(new_frame)
                     queue.task_done()
                     new_frame = await queue.get()
 
+            return output_frames
+
         if direction == FrameDirection.UPSTREAM:
             # If we get an upstream frame we process it in each sink.
-            await asyncio.gather(
+            frames_per_pipeline = await asyncio.gather(
                 *[wait_for_sync(s, self._up_queue, frame, direction) for s in self._sinks]
             )
         elif direction == FrameDirection.DOWNSTREAM:
             # If we get a downstream frame we process it in each source.
-            await asyncio.gather(
+            frames_per_pipeline = await asyncio.gather(
                 *[wait_for_sync(s, self._down_queue, frame, direction) for s in self._sources]
             )
 
-        seen_ids = set()
-        while not self._up_queue.empty():
-            frame = await self._up_queue.get()
-            if frame.id not in seen_ids:
-                await self.push_frame(frame, FrameDirection.UPSTREAM)
-                seen_ids.add(frame.id)
-            self._up_queue.task_done()
+        if use_pipeline_order:
+            # Push frames in pipeline definition order, deduplicating by id.
+            seen_ids = set()
+            for pipeline_frames in frames_per_pipeline:
+                for f in pipeline_frames:
+                    if f.id not in seen_ids:
+                        await self.push_frame(f, direction)
+                        seen_ids.add(f.id)
+        else:
+            # ARRIVAL mode: drain the shared queues in the order frames arrived.
+            seen_ids = set()
+            while not self._up_queue.empty():
+                frame = await self._up_queue.get()
+                if frame.id not in seen_ids:
+                    await self.push_frame(frame, FrameDirection.UPSTREAM)
+                    seen_ids.add(frame.id)
+                self._up_queue.task_done()
 
-        seen_ids = set()
-        while not self._down_queue.empty():
-            frame = await self._down_queue.get()
-            if frame.id not in seen_ids:
-                await self.push_frame(frame, FrameDirection.DOWNSTREAM)
-                seen_ids.add(frame.id)
-            self._down_queue.task_done()
+            seen_ids = set()
+            while not self._down_queue.empty():
+                frame = await self._down_queue.get()
+                if frame.id not in seen_ids:
+                    await self.push_frame(frame, FrameDirection.DOWNSTREAM)
+                    seen_ids.add(frame.id)
+                self._down_queue.task_done()

src/pipecat/processors/aggregators/llm_response_universal.py

@@ -75,6 +75,7 @@ from pipecat.processors.aggregators.llm_context_summarizer import (
     SummaryAppliedEvent,
 )
 from pipecat.processors.frame_processor import FrameCallback, FrameDirection, FrameProcessor
+from pipecat.services.settings import LLMSettings
 from pipecat.turns.user_idle_controller import UserIdleController
 from pipecat.turns.user_mute import BaseUserMuteStrategy
 from pipecat.turns.user_start import BaseUserTurnStartStrategy, UserTurnStartedParams
@@ -446,6 +447,9 @@ class LLMUserAggregator(LLMContextAggregator):
         self._user_turn_controller.add_event_handler(
             "on_user_turn_stop_timeout", self._on_user_turn_stop_timeout
         )
+        self._user_turn_controller.add_event_handler(
+            "on_reset_aggregation", self._on_reset_aggregation
+        )
 
         self._user_idle_controller = UserIdleController(
             user_idle_timeout=self._params.user_idle_timeout
@@ -561,10 +565,10 @@ class LLMUserAggregator(LLMContextAggregator):
             # Enable the feature on the LLM with config
             await self.push_frame(
                 LLMUpdateSettingsFrame(
-                    settings={
-                        "filter_incomplete_user_turns": True,
-                        "user_turn_completion_config": config,
-                    }
+                    delta=LLMSettings(
+                        filter_incomplete_user_turns=True,
+                        user_turn_completion_config=config,
+                    )
                 )
             )
 
@@ -747,6 +751,12 @@ class LLMUserAggregator(LLMContextAggregator):
 
         await self._maybe_emit_user_turn_stopped(strategy)
 
+    async def _on_reset_aggregation(
+        self, controller: UserTurnController, strategy: BaseUserTurnStartStrategy
+    ):
+        logger.debug(f"{self}: Resetting aggregation (strategy: {strategy})")
+        await self.reset()
+
     async def _on_user_turn_stop_timeout(self, controller):
         await self._call_event_handler("on_user_turn_stop_timeout")
 

src/pipecat/processors/filters/wake_check_filter.py

@@ -6,6 +6,9 @@
 
 """Wake phrase detection filter for Pipecat transcription processing.
 
+.. deprecated:: 0.0.106
+    Use :class:`~pipecat.turns.user_start.WakePhraseUserTurnStartStrategy` instead.
+
 This module provides a frame processor that filters transcription frames,
 only allowing them through after wake phrases have been detected. Includes
 keepalive functionality to maintain conversation flow after wake detection.
@@ -13,6 +16,7 @@ keepalive functionality to maintain conversation flow after wake detection.
 
 import re
 import time
+import warnings
 from enum import Enum
 from typing import List
 
@@ -25,6 +29,11 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 class WakeCheckFilter(FrameProcessor):
     """Frame processor that filters transcription frames based on wake phrase detection.
 
+    .. deprecated:: 0.0.106
+        Use :class:`~pipecat.turns.user_start.WakePhraseUserTurnStartStrategy` instead,
+        which integrates with the user turn strategy system and supports configurable
+        timeouts and single-activation mode.
+
     This filter monitors transcription frames for configured wake phrases and only
     passes frames through after a wake phrase has been detected. Maintains a
     keepalive timeout to allow continued conversation after wake detection.
@@ -65,12 +74,21 @@ class WakeCheckFilter(FrameProcessor):
     def __init__(self, wake_phrases: List[str], keepalive_timeout: float = 3):
         """Initialize the wake phrase filter.
 
+        .. deprecated:: 0.0.106
+            Use :class:`~pipecat.turns.user_start.WakePhraseUserTurnStartStrategy` instead.
+
         Args:
             wake_phrases: List of wake phrases to detect in transcriptions.
             keepalive_timeout: Duration in seconds to keep passing frames after
                 wake detection. Defaults to 3 seconds.
         """
         super().__init__()
+        warnings.warn(
+            "WakeCheckFilter is deprecated since v0.0.106. "
+            "Use WakePhraseUserTurnStartStrategy instead.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
         self._participant_states = {}
         self._keepalive_timeout = keepalive_timeout
         self._wake_patterns = []

src/pipecat/runner/daily.py

@@ -79,16 +79,17 @@ async def configure(
     aiohttp_session: aiohttp.ClientSession,
     *,
     api_key: Optional[str] = None,
-    room_exp_duration: Optional[float] = 2.0,
-    token_exp_duration: Optional[float] = 2.0,
+    room_exp_duration: float = 2.0,
+    token_exp_duration: float = 2.0,
     sip_caller_phone: Optional[str] = None,
-    sip_enable_video: Optional[bool] = False,
-    sip_num_endpoints: Optional[int] = 1,
+    sip_enable_video: bool = False,
+    sip_num_endpoints: int = 1,
+    enable_dialout: bool = False,
     sip_codecs: Optional[Dict[str, List[str]]] = None,
     sip_provider: Optional[str] = None,
     room_geo: Optional[str] = None,
     room_properties: Optional[DailyRoomProperties] = None,
-    token_properties: Optional["DailyMeetingTokenProperties"] = None,
+    token_properties: Optional[DailyMeetingTokenProperties] = None,
 ) -> DailyRoomConfig:
     """Configure Daily room URL and token with optional SIP capabilities.
 
@@ -105,6 +106,8 @@ async def configure(
             When provided, enables SIP functionality and returns SipRoomConfig.
         sip_enable_video: Whether video is enabled for SIP.
         sip_num_endpoints: Number of allowed SIP endpoints.
+        enable_dialout: Whether to enable outbound dialing (PSTN or SIP) on the room.
+            Requires dial-out entitlement on your Daily account.
         sip_codecs: Codecs to support for audio and video. If None, uses Daily defaults.
             Example: {"audio": ["OPUS"], "video": ["H264"]}
         sip_provider: SIP provider name (e.g., "daily"). Only used when
@@ -159,6 +162,7 @@ async def configure(
                 sip_caller_phone is not None,
                 sip_enable_video is not False,
                 sip_num_endpoints != 1,
+                enable_dialout is not False,
                 sip_codecs is not None,
                 sip_provider is not None,
                 room_geo is not None,
@@ -184,6 +188,8 @@ async def configure(
         aiohttp_session=aiohttp_session,
     )
 
+    token_expiry_seconds: float = token_exp_duration * 60 * 60
+
     # Check for existing room URL (only in standard mode)
     existing_room_url = os.getenv("DAILY_ROOM_URL")
     if existing_room_url and not sip_enabled:
@@ -192,15 +198,16 @@ async def configure(
         room_url = existing_room_url
 
         # Create token and return standard format
-        expiry_time: float = token_exp_duration * 60 * 60
         token_params = None
         if token_properties:
             token_params = DailyMeetingTokenParams(properties=token_properties)
-        token = await daily_rest_helper.get_token(room_url, expiry_time, params=token_params)
+        token = await daily_rest_helper.get_token(
+            room_url, token_expiry_seconds, params=token_params
+        )
         return DailyRoomConfig(room_url=room_url, token=token)
 
     # Create a new room
-    room_prefix = "pipecat-sip" if sip_enabled else "pipecat"
+    room_prefix = "pipecat-telephony" if (sip_enabled or enable_dialout) else "pipecat"
     room_name = f"{room_prefix}-{uuid.uuid4().hex[:8]}"
     logger.info(f"Creating new Daily room: {room_name}")
 
@@ -218,6 +225,9 @@ async def configure(
         if room_geo:
             room_properties.geo = room_geo
 
+        if enable_dialout:
+            room_properties.enable_dialout = True
+
         # Add SIP configuration if enabled
         if sip_enabled:
             sip_params = DailyRoomSipParams(
@@ -229,7 +239,6 @@ async def configure(
                 provider=sip_provider,
             )
             room_properties.sip = sip_params
-            room_properties.enable_dialout = True  # Enable outbound calls if needed
             room_properties.start_video_off = not sip_enable_video  # Voice-only by default
 
     # Create room parameters
@@ -241,7 +250,6 @@ async def configure(
         logger.info(f"Created Daily room: {room_url}")
 
         # Create meeting token
-        token_expiry_seconds = token_exp_duration * 60 * 60
         token_params = None
         if token_properties:
             token_params = DailyMeetingTokenParams(properties=token_properties)

src/pipecat/serializers/genesys.py

@@ -336,8 +336,7 @@ class GenesysAudioHookSerializer(FrameSerializer):
         if include_position:
             msg["position"] = self._format_position(self._position)
 
-        if parameters:
-            msg["parameters"] = parameters
+        msg["parameters"] = parameters if parameters is not None else {}
 
         return msg
 

src/pipecat/services/aws/nova_sonic/llm.py

@@ -27,8 +27,8 @@ from pydantic import BaseModel, Field
 from pipecat.adapters.schemas.tools_schema import ToolsSchema
 from pipecat.adapters.services.aws_nova_sonic_adapter import AWSNovaSonicLLMAdapter, Role
 from pipecat.frames.frames import (
+    AggregatedTextFrame,
     AggregationType,
-    BotStoppedSpeakingFrame,
     CancelFrame,
     EndFrame,
     Frame,
@@ -424,18 +424,16 @@ class AWSNovaSonicLLMService(LLMService):
         self._input_audio_content_name: Optional[str] = None
         self._content_being_received: Optional[CurrentContent] = None
         self._assistant_is_responding = False
-        self._may_need_repush_assistant_text = False
         self._ready_to_send_context = False
-        self._handling_bot_stopped_speaking = False
         self._triggering_assistant_response = False
         self._waiting_for_trigger_transcription = False
         self._disconnecting = False
         self._connected_time: Optional[float] = None
         self._wants_connection = False
         self._user_text_buffer = ""
-        self._assistant_text_buffer = ""
         self._completed_tool_calls = set()
         self._audio_input_started = False
+        self._pending_speculative_text: Optional[str] = None
 
         file_path = files("pipecat.services.aws.nova_sonic").joinpath("ready.wav")
         with wave.open(file_path.open("rb"), "rb") as wav_file:
@@ -505,11 +503,13 @@ class AWSNovaSonicLLMService(LLMService):
     async def reset_conversation(self):
         """Reset the conversation state while preserving context.
 
-        Handles bot stopped speaking event, disconnects from the service,
-        and reconnects with the preserved context.
+        Cleans up any in-progress assistant response, disconnects from the
+        service, and reconnects with the preserved context.
         """
         logger.debug("Resetting conversation")
-        await self._handle_bot_stopped_speaking(delay_to_catch_trailing_assistant_text=False)
+        if self._assistant_is_responding:
+            self._assistant_is_responding = False
+            await self._report_assistant_response_ended()
 
         # Grab context to carry through disconnect/reconnect
         context = self._context
@@ -540,8 +540,6 @@ class AWSNovaSonicLLMService(LLMService):
             await self._handle_context(context)
         elif isinstance(frame, InputAudioRawFrame):
             await self._handle_input_audio_frame(frame)
-        elif isinstance(frame, BotStoppedSpeakingFrame):
-            await self._handle_bot_stopped_speaking(delay_to_catch_trailing_assistant_text=True)
         elif isinstance(frame, InterruptionFrame):
             await self._handle_interruption_frame()
 
@@ -569,49 +567,8 @@ class AWSNovaSonicLLMService(LLMService):
 
         await self._send_user_audio_event(frame.audio)
 
-    async def _handle_bot_stopped_speaking(self, delay_to_catch_trailing_assistant_text: bool):
-        # Protect against back-to-back BotStoppedSpeaking calls, which I've observed
-        if self._handling_bot_stopped_speaking:
-            return
-        self._handling_bot_stopped_speaking = True
-
-        async def finalize_assistant_response():
-            if self._assistant_is_responding:
-                # Consider the assistant finished with their response (possibly after a short delay,
-                # to allow for any trailing FINAL assistant text block to come in that need to make
-                # it into context).
-                #
-                # TODO: ideally we could base this solely on the LLM output events, but I couldn't
-                # figure out a reliable way to determine when we've gotten our last FINAL text block
-                # after the LLM is done talking.
-                #
-                # First I looked at stopReason, but it doesn't seem like the last FINAL text block
-                # is reliably marked END_TURN (sometimes the *first* one is, but not the last...
-                # bug?)
-                #
-                # Then I considered schemes where we tally or match up SPECULATIVE text blocks with
-                # FINAL text blocks to know how many or which FINAL blocks to expect, but user
-                # interruptions throw a wrench in these schemes: depending on the exact timing of
-                # the interruption, we should or shouldn't expect some FINAL blocks.
-                if delay_to_catch_trailing_assistant_text:
-                    # This delay length is a balancing act between "catching" trailing assistant
-                    # text that is quite delayed but not waiting so long that user text comes in
-                    # first and results in a bit of context message order scrambling.
-                    await asyncio.sleep(1.25)
-                self._assistant_is_responding = False
-                await self._report_assistant_response_ended()
-
-            self._handling_bot_stopped_speaking = False
-
-        # Finalize the assistant response, either now or after a delay
-        if delay_to_catch_trailing_assistant_text:
-            self.create_task(finalize_assistant_response())
-        else:
-            await finalize_assistant_response()
-
     async def _handle_interruption_frame(self):
-        if self._assistant_is_responding:
-            self._may_need_repush_assistant_text = True
+        pass
 
     #
     # LLM communication: lifecycle
@@ -771,17 +728,15 @@ class AWSNovaSonicLLMService(LLMService):
             self._input_audio_content_name = None
             self._content_being_received = None
             self._assistant_is_responding = False
-            self._may_need_repush_assistant_text = False
             self._ready_to_send_context = False
-            self._handling_bot_stopped_speaking = False
             self._triggering_assistant_response = False
             self._waiting_for_trigger_transcription = False
             self._disconnecting = False
             self._connected_time = None
             self._user_text_buffer = ""
-            self._assistant_text_buffer = ""
             self._completed_tool_calls = set()
             self._audio_input_started = False
+            self._pending_speculative_text = None
 
             logger.info("Finished disconnecting")
         except Exception as e:
@@ -1153,10 +1108,11 @@ class AWSNovaSonicLLMService(LLMService):
         self._content_being_received = content
 
         if content.role == Role.ASSISTANT:
-            if content.type == ContentType.AUDIO:
-                # Note that an assistant response can comprise of multiple audio blocks
-                if not self._assistant_is_responding:
-                    # The assistant has started responding.
+            if content.type == ContentType.TEXT:
+                if (
+                    content.text_stage == TextStage.SPECULATIVE
+                    and not self._assistant_is_responding
+                ):
                     self._assistant_is_responding = True
                     await self._report_user_transcription_ended()  # Consider user turn over
                     await self._report_assistant_response_started()
@@ -1232,18 +1188,30 @@ class AWSNovaSonicLLMService(LLMService):
 
         if content.role == Role.ASSISTANT:
             if content.type == ContentType.TEXT:
-                # Ignore non-final text, and the "interrupted" message (which isn't meaningful text)
-                if content.text_stage == TextStage.FINAL and stop_reason != "INTERRUPTED":
-                    if self._assistant_is_responding:
-                        # Text added to the ongoing assistant response
-                        await self._report_assistant_response_text_added(content.text_content)
+                if stop_reason != "INTERRUPTED":
+                    if content.text_stage == TextStage.SPECULATIVE:
+                        await self._report_llm_text(content.text_content)
+                elif self._assistant_is_responding:
+                    # TEXT INTERRUPTED with no audio means the user interrupted
+                    # before audio started. End the response here since no AUDIO
+                    # contentEnd will arrive.
+                    self._assistant_is_responding = False
+                    await self._report_assistant_response_ended()
+            elif content.type == ContentType.AUDIO:
+                # Emit deferred TTSTextFrame after all audio chunks have been sent
+                await self._report_tts_text()
+                if stop_reason in ("END_TURN", "INTERRUPTED"):
+                    # END_TURN: normal completion. INTERRUPTED: user interrupted
+                    # mid-audio. Both mean no more audio for this turn.
+                    self._assistant_is_responding = False
+                    await self._report_assistant_response_ended()
         elif content.role == Role.USER:
             if content.type == ContentType.TEXT:
                 if content.text_stage == TextStage.FINAL:
                     # User transcription text added
                     await self._report_user_transcription_text_added(content.text_content)
 
-    async def _handle_completion_end_event(self, event_json):
+    async def _handle_completion_end_event(self, _):
         pass
 
     #
@@ -1256,29 +1224,40 @@ class AWSNovaSonicLLMService(LLMService):
 
     async def _report_assistant_response_started(self):
         logger.debug("Assistant response started")
-
-        # Report the start of the assistant response.
         await self.push_frame(LLMFullResponseStartFrame())
 
         # Report that equivalent of TTS (this is a speech-to-speech model) started
         await self.push_frame(TTSStartedFrame())
 
-    async def _report_assistant_response_text_added(self, text):
-        if not self._context:  # should never happen
-            return
+    async def _report_llm_text(self, text):
+        """Push speculative assistant text and defer TTSTextFrame.
 
-        logger.debug(f"Assistant response text added: {text}")
+        Speculative text arrives before each audio chunk, providing real-time
+        text that is synchronized with what the bot is saying. LLMTextFrame and
+        AggregatedTextFrame are pushed immediately for real-time text display.
+        TTSTextFrame emission is deferred to audio contentEnd so it aligns with
+        audio playout timing.
+        """
+        logger.debug(f"Assistant speculative text: {text}")
 
-        # Report the text of the assistant response.
-        await self._push_assistant_response_text_frames(text)
+        llm_text_frame = LLMTextFrame(text)
+        llm_text_frame.append_to_context = False
+        await self.push_frame(llm_text_frame)
 
-        # HACK: here we're also buffering the assistant text ourselves as a
-        # backup rather than relying solely on the assistant context aggregator
-        # to do it, because the text arrives from Nova Sonic only after all the
-        # assistant audio frames have been pushed, meaning that if an
-        # interruption frame were to arrive we would lose all of it (the text
-        # frames sitting in the queue would be wiped).
-        self._assistant_text_buffer += text
+        aggregated_text_frame = AggregatedTextFrame(text, aggregated_by=AggregationType.SENTENCE)
+        aggregated_text_frame.append_to_context = False
+        await self.push_frame(aggregated_text_frame)
+
+        self._pending_speculative_text = text
+
+    async def _report_tts_text(self):
+        if self._pending_speculative_text:
+            tts_text_frame = TTSTextFrame(
+                self._pending_speculative_text, aggregated_by=AggregationType.SENTENCE
+            )
+            tts_text_frame.includes_inter_frame_spaces = True
+            await self.push_frame(tts_text_frame)
+            self._pending_speculative_text = None
 
     async def _report_assistant_response_ended(self):
         if not self._context:  # should never happen
@@ -1286,54 +1265,12 @@ class AWSNovaSonicLLMService(LLMService):
 
         logger.debug("Assistant response ended")
 
-        # If an interruption frame arrived while the assistant was responding
-        # we may have lost all of the assistant text (see HACK, above), so
-        # re-push it downstream to the aggregator now.
-        if self._may_need_repush_assistant_text:
-            # Just in case, check that assistant text hasn't already made it
-            # into the context (sometimes it does, despite the interruption).
-            messages = self._context.get_messages()
-            last_message = messages[-1] if messages else None
-            if (
-                not last_message
-                or last_message.get("role") != "assistant"
-                or last_message.get("content") != self._assistant_text_buffer
-            ):
-                # We also need to re-push the LLMFullResponseStartFrame since the
-                # TTSTextFrame would be ignored otherwise (the interruption frame
-                # would have cleared the assistant aggregator state).
-                await self.push_frame(LLMFullResponseStartFrame())
-                await self._push_assistant_response_text_frames(self._assistant_text_buffer)
-            self._may_need_repush_assistant_text = False
-
         # Report the end of the assistant response.
         await self.push_frame(LLMFullResponseEndFrame())
 
         # Report that equivalent of TTS (this is a speech-to-speech model) stopped.
         await self.push_frame(TTSStoppedFrame())
 
-        # Clear out the buffered assistant text
-        self._assistant_text_buffer = ""
-
-    async def _push_assistant_response_text_frames(self, text: str):
-        # In a typical "cascade" LLM + TTS setup, LLMTextFrames would not
-        # proceed beyond the TTS service. Therefore, since a speech-to-speech
-        # service like Nova Sonic combines both LLM and TTS functionality, you
-        # would think we wouldn't need to push LLMTextFrames at all. However,
-        # RTVI relies on LLMTextFrames being pushed to trigger its
-        # "bot-llm-text" event. So here we push an LLMTextFrame, too, but avoid
-        # appending it to context to avoid context message duplication.
-
-        # Push LLMTextFrame
-        llm_text_frame = LLMTextFrame(text)
-        llm_text_frame.append_to_context = False
-        await self.push_frame(llm_text_frame)
-
-        # Push TTSTextFrame
-        tts_text_frame = TTSTextFrame(text, aggregated_by=AggregationType.SENTENCE)
-        tts_text_frame.includes_inter_frame_spaces = True
-        await self.push_frame(tts_text_frame)
-
     #
     # user transcription reporting
     #
@@ -1363,6 +1300,12 @@ class AWSNovaSonicLLMService(LLMService):
         if not self._context:  # should never happen
             return
 
+        # Nothing to report if no user speech was transcribed (e.g. the prompt
+        # was text-only, which is the case on the first user turn when the bot
+        # starts the conversation).
+        if not self._user_text_buffer:
+            return
+
         logger.debug(f"User transcription ended")
 
         # Report to the upstream user context aggregator that some new user

src/pipecat/services/deepgram/sagemaker/tts.py

@@ -20,18 +20,13 @@ from typing import Any, AsyncGenerator, Optional
 from loguru import logger
 
 from pipecat.frames.frames import (
-    BotStoppedSpeakingFrame,
     CancelFrame,
     EndFrame,
     ErrorFrame,
     Frame,
-    InterruptionFrame,
-    LLMFullResponseEndFrame,
     StartFrame,
     TTSAudioRawFrame,
-    TTSStartedFrame,
 )
-from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.aws.sagemaker.bidi_client import SageMakerBidiClient
 from pipecat.services.settings import TTSSettings
 from pipecat.services.tts_service import TTSService
@@ -115,6 +110,7 @@ class DeepgramSageMakerTTSService(TTSService):
 
         super().__init__(
             sample_rate=sample_rate,
+            push_start_frame=True,
             push_stop_frames=True,
             pause_frame_processing=True,
             append_trailing_space=True,
@@ -128,8 +124,6 @@ class DeepgramSageMakerTTSService(TTSService):
 
         self._client: Optional[SageMakerBidiClient] = None
         self._response_task: Optional[asyncio.Task] = None
-        self._context_id: Optional[str] = None
-        self._ttfb_started: bool = False
 
     def can_generate_metrics(self) -> bool:
         """Check if this service can generate processing metrics.
@@ -166,20 +160,6 @@ class DeepgramSageMakerTTSService(TTSService):
         await super().cancel(frame)
         await self._disconnect()
 
-    async def process_frame(self, frame: Frame, direction: FrameDirection):
-        """Process frames with special handling for LLM response end.
-
-        Args:
-            frame: The frame to process.
-            direction: The direction of frame processing.
-        """
-        await super().process_frame(frame, direction)
-
-        if isinstance(frame, (LLMFullResponseEndFrame, EndFrame)):
-            await self.flush_audio()
-        elif isinstance(frame, BotStoppedSpeakingFrame):
-            self._ttfb_started = False
-
     async def _connect(self):
         """Connect to the SageMaker endpoint and start the BiDi session.
 
@@ -301,13 +281,14 @@ class DeepgramSageMakerTTSService(TTSService):
                         except (UnicodeDecodeError, json.JSONDecodeError):
                             # Not JSON — treat as raw audio bytes
                             await self.stop_ttfb_metrics()
+                            context_id = self.get_active_audio_context_id()
                             frame = TTSAudioRawFrame(
                                 payload,
                                 self.sample_rate,
                                 1,
-                                context_id=self._context_id,
+                                context_id=context_id,
                             )
-                            await self.push_frame(frame)
+                            await self.append_to_audio_context(context_id, frame)
 
         except asyncio.CancelledError:
             logger.debug("TTS response processor cancelled")
@@ -316,15 +297,13 @@ class DeepgramSageMakerTTSService(TTSService):
         finally:
             logger.debug("TTS response processor stopped")
 
-    async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection):
-        """Handle interruption by sending Clear message to Deepgram.
+    async def on_audio_context_interrupted(self, context_id: str):
+        """Called when an audio context is cancelled due to an interruption.
 
-        The Clear message will clear Deepgram's internal text buffer and stop
-        sending audio, allowing for a new response to be generated.
+        Args:
+            context_id: The ID of the audio context that was interrupted, or
+                ``None`` if no context was active at the time.
         """
-        await super()._handle_interruption(frame, direction)
-        self._ttfb_started = False
-
         if self._client and self._client.is_active:
             try:
                 await self._client.send_json({"type": "Clear"})
@@ -356,19 +335,8 @@ class DeepgramSageMakerTTSService(TTSService):
             the response processor).
         """
         logger.debug(f"{self}: Generating TTS [{text}]")
-
         try:
-            if not self.audio_context_available(context_id):
-                await self.create_audio_context(context_id)
-                if not self._ttfb_started:
-                    await self.start_ttfb_metrics()
-                    self._ttfb_started = True
-                yield TTSStartedFrame(context_id=context_id)
-            self._context_id = context_id
-
             await self._client.send_json({"type": "Speak", "text": text})
-
             yield None
-
         except Exception as e:
             yield ErrorFrame(error=f"Unknown error occurred: {e}")

src/pipecat/services/deepgram/stt.py

@@ -247,6 +247,45 @@ class DeepgramSTTSettings(STTSettings):
                 del self.extra[key]
 
 
+def _derive_deepgram_urls(base_url: str) -> tuple[str, str]:
+    """Derive paired WebSocket and HTTP URLs from a single base URL.
+
+    The Deepgram SDK client requires both a WebSocket URL (for streaming)
+    and an HTTP URL (for REST calls). This helper lets developers provide
+    a single ``base_url`` and consistently derives both, preserving the
+    security level they chose. Useful for air-gapped or private deployments
+    where insecure schemes (ws:// / http://) are acceptable.
+
+    Accepted inputs:
+        - ``wss://`` or ``https://`` — secure (paired as wss + https)
+        - ``ws://`` or ``http://`` — insecure (paired as ws + http)
+        - Bare hostname (no scheme) — defaults to secure
+        - Unrecognized scheme — logs a warning, defaults to secure
+
+    Args:
+        base_url: Host with optional scheme, port, and path.
+
+    Returns:
+        A (ws_url, http_url) tuple with consistent schemes.
+    """
+    known_schemes = ("wss://", "https://", "ws://", "http://")
+    if "://" in base_url:
+        scheme, host = base_url.split("://", 1)
+        scheme += "://"
+        if scheme not in known_schemes:
+            logger.warning(
+                f"Unrecognized scheme in base_url '{base_url}', defaulting to wss:// / https://"
+            )
+    else:
+        scheme = ""
+        host = base_url
+
+    insecure = scheme in ("ws://", "http://")
+    ws_url = f"{'ws' if insecure else 'wss'}://{host}"
+    http_url = f"{'http' if insecure else 'https'}://{host}"
+    return ws_url, http_url
+
+
 class DeepgramSTTService(STTService):
     """Deepgram speech-to-text service.
 
@@ -445,8 +484,7 @@ class DeepgramSTTService(STTService):
             try:
                 from deepgram import DeepgramClientEnvironment
 
-                ws_url = base_url if base_url.startswith("wss://") else f"wss://{base_url}"
-                http_url = base_url if base_url.startswith("https://") else f"https://{base_url}"
+                ws_url, http_url = _derive_deepgram_urls(base_url)
                 environment = DeepgramClientEnvironment(
                     base=http_url,
                     production=ws_url,
@@ -554,7 +592,15 @@ class DeepgramSTTService(STTService):
             value = getattr(s, f.name)
             if not is_given(value) or value is None:
                 continue
-            kwargs[f.name] = str(value).lower() if isinstance(value, bool) else str(value)
+            # Lists (e.g. keyterm, keywords, search, redact, replace) must be
+            # passed through as-is so the SDK's encode_query produces repeated
+            # query params (keyterm=a&keyterm=b) instead of a stringified list.
+            if isinstance(value, list):
+                kwargs[f.name] = value
+            elif isinstance(value, bool):
+                kwargs[f.name] = str(value).lower()
+            else:
+                kwargs[f.name] = str(value)
 
         # model and language
         if is_given(s.model) and s.model is not None:
@@ -580,7 +626,12 @@ class DeepgramSTTService(STTService):
         # Any remaining values in extra (that didn't map to declared fields)
         for key, value in s.extra.items():
             if value is not None:
-                kwargs[key] = str(value).lower() if isinstance(value, bool) else str(value)
+                if isinstance(value, list):
+                    kwargs[key] = value
+                elif isinstance(value, bool):
+                    kwargs[key] = str(value).lower()
+                else:
+                    kwargs[key] = str(value)
 
         if self._addons:
             for key, value in self._addons.items():

src/pipecat/services/fish/tts.py

@@ -21,12 +21,10 @@ from pipecat.frames.frames import (
     EndFrame,
     ErrorFrame,
     Frame,
-    InterruptionFrame,
     StartFrame,
     TTSAudioRawFrame,
     TTSStoppedFrame,
 )
-from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.settings import NOT_GIVEN, TTSSettings, _NotGiven
 from pipecat.services.tts_service import InterruptibleTTSService
 from pipecat.transcriptions.language import Language
@@ -362,8 +360,8 @@ class FishAudioTTSService(InterruptibleTTSService):
             return self._websocket
         raise Exception("Websocket not connected")
 
-    async def _handle_interruption(self, frame: InterruptionFrame, direction: FrameDirection):
-        await super()._handle_interruption(frame, direction)
+    async def on_audio_context_interrupted(self, context_id: str):
+        """Stop all metrics when audio context is interrupted."""
         await self.stop_all_metrics()
 
     async def _receive_messages(self):
@@ -377,8 +375,14 @@ class FishAudioTTSService(InterruptibleTTSService):
                             audio_data = msg.get("audio")
                             # Only process larger chunks to remove msgpack overhead
                             if audio_data and len(audio_data) > 1024:
-                                frame = TTSAudioRawFrame(audio_data, self.sample_rate, 1)
-                                await self.push_frame(frame)
+                                context_id = self.get_active_audio_context_id()
+                                frame = TTSAudioRawFrame(
+                                    audio_data,
+                                    self.sample_rate,
+                                    1,
+                                    context_id=context_id,
+                                )
+                                await self.append_to_audio_context(context_id, frame)
                                 await self.stop_ttfb_metrics()
                         elif event == "finish":
                             reason = msg.get("reason", "unknown")

src/pipecat/services/gradium/stt.py

@@ -10,6 +10,7 @@ This module provides integration with Gradium's real-time speech-to-text
 WebSocket API for streaming audio transcription.
 """
 
+import asyncio
 import base64
 import json
 from dataclasses import dataclass, field
@@ -22,6 +23,7 @@ from pipecat.frames.frames import (
     CancelFrame,
     EndFrame,
     Frame,
+    InterimTranscriptionFrame,
     StartFrame,
     TranscriptionFrame,
     VADUserStartedSpeakingFrame,
@@ -43,7 +45,37 @@ except ModuleNotFoundError as e:
     logger.error('In order to use Gradium, you need to `pip install "pipecat-ai[gradium]"`.')
     raise Exception(f"Missing module: {e}")
 
-SAMPLE_RATE = 24000
+# Seconds to wait after a "flushed" message for trailing text tokens to arrive
+# before finalizing the transcription.
+TRANSCRIPT_AGGREGATION_DELAY = 0.1
+
+
+def _input_format_from_encoding(encoding: str, sample_rate: int) -> str:
+    """Build Gradium input_format from encoding type and sample rate.
+
+    For PCM encoding, appends the sample rate (e.g., "pcm_16000").
+    For other encodings (wav, opus), returns the encoding as-is.
+
+    Args:
+        encoding: Base encoding type ("pcm", "wav", or "opus").
+        sample_rate: Audio sample rate in Hz.
+
+    Returns:
+        The full input_format string for the Gradium API.
+    """
+    if encoding == "pcm":
+        match sample_rate:
+            case 8000:
+                return "pcm_8000"
+            case 16000:
+                return "pcm_16000"
+            case 24000:
+                return "pcm_24000"
+        logger.warning(
+            f"GradiumSTTService: unsupported sample rate {sample_rate} for PCM encoding, using pcm_16000"
+        )
+        return "pcm_16000"
+    return encoding
 
 
 def language_to_gradium_language(language: Language) -> Optional[str]:
@@ -115,6 +147,8 @@ class GradiumSTTService(WebsocketSTTService):
         *,
         api_key: str,
         api_endpoint_base_url: str = "wss://eu.api.gradium.ai/api/speech/asr",
+        encoding: str = "pcm",
+        sample_rate: Optional[int] = None,
         params: Optional[InputParams] = None,
         json_config: Optional[str] = None,
         settings: Optional[Settings] = None,
@@ -126,6 +160,12 @@ class GradiumSTTService(WebsocketSTTService):
         Args:
             api_key: Gradium API key for authentication.
             api_endpoint_base_url: WebSocket endpoint URL. Defaults to Gradium's streaming endpoint.
+            encoding: Base audio encoding type. One of "pcm", "wav", or "opus".
+                For PCM, the sample rate is appended automatically from the
+                pipeline's audio_in_sample_rate (e.g., "pcm" becomes "pcm_16000").
+                Defaults to "pcm".
+            sample_rate: Audio sample rate in Hz. If None, uses the pipeline
+                sample rate.
             params: Configuration parameters for language and delay settings.
 
                 .. deprecated:: 0.0.105
@@ -153,7 +193,7 @@ class GradiumSTTService(WebsocketSTTService):
 
         # 1. Initialize default_settings with hardcoded defaults
         default_settings = self.Settings(
-            model=None,
+            model="default",
             language=None,
             delay_in_frames=None,
         )
@@ -173,7 +213,7 @@ class GradiumSTTService(WebsocketSTTService):
             default_settings.apply_update(settings)
 
         super().__init__(
-            sample_rate=SAMPLE_RATE,
+            sample_rate=sample_rate,
             ttfs_p99_latency=ttfs_p99_latency,
             settings=default_settings,
             **kwargs,
@@ -181,19 +221,25 @@ class GradiumSTTService(WebsocketSTTService):
 
         self._api_key = api_key
         self._api_endpoint_base_url = api_endpoint_base_url
+        self._encoding = encoding
         self._websocket = None
         self._json_config = json_config
 
         self._receive_task = None
 
+        self._input_format = ""
+
         self._audio_buffer = bytearray()
         self._chunk_size_ms = 80
         self._chunk_size_bytes = 0
 
-        # Set from the ready message when connecting to the service.
-        # These values are used for flushing transcription.
-        self._delay_in_frames = 0
-        self._frame_size = 0
+        # Accumulates text fragments within a turn. Each "text" message
+        # appends to this list. On "flushed" a short aggregation delay
+        # allows trailing tokens to arrive before the full text is joined
+        # and pushed as a TranscriptionFrame.
+        self._accumulated_text: list[str] = []
+        self._flush_counter = 0
+        self._transcript_aggregation_task: Optional[asyncio.Task] = None
 
     def can_generate_metrics(self) -> bool:
         """Check if the service can generate metrics.
@@ -228,6 +274,7 @@ class GradiumSTTService(WebsocketSTTService):
             frame: Start frame to begin processing.
         """
         await super().start(frame)
+        self._input_format = _input_format_from_encoding(self._encoding, self.sample_rate)
         self._chunk_size_bytes = int(self._chunk_size_ms * self.sample_rate * 2 / 1000)
         await self._connect()
 
@@ -249,56 +296,41 @@ class GradiumSTTService(WebsocketSTTService):
         await super().cancel(frame)
         await self._disconnect()
 
-    async def process_frame(self, frame: Frame, direction: FrameDirection):
-        """Process frames with VAD-specific handling.
+    async def _start_metrics(self):
+        """Start performance metrics collection for transcription processing."""
+        await self.start_processing_metrics()
 
-        When VAD detects the user has stopped speaking, we flush the transcription
-        by sending silence frames. This makes the system more reactive by getting
-        the final transcription faster without closing the connection.
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        """Process incoming frames and handle speech events.
 
         Args:
             frame: The frame to process.
-            direction: The direction of frame processing.
+            direction: Direction of frame flow in the pipeline.
         """
         await super().process_frame(frame, direction)
 
         if isinstance(frame, VADUserStartedSpeakingFrame):
-            await self.start_processing_metrics()
+            await self._start_metrics()
         elif isinstance(frame, VADUserStoppedSpeakingFrame):
-            await self._flush_transcription()
+            await self._send_flush()
 
-    async def _flush_transcription(self):
-        """Flush the transcription by sending silence frames.
+    async def _send_flush(self):
+        """Send a flush request to process any buffered audio immediately.
 
-        When VAD detects the user stopped speaking, we send delay_in_frames
-        chunks of silence (zeros) to flush the remaining audio from the model's
-        buffer. This allows for faster turn-around without closing the connection.
-
-        From Gradium docs: "feed in delay_in_frames chunks of silence (vectors
-        of zeros). If those are fed in faster than realtime, the API also has
-        a possibility to process them faster."
+        Sends a flush message to tell the server to process buffered audio.
+        The server responds with text fragments followed by a "flushed"
+        acknowledgment, which triggers finalization.
         """
         if not self._websocket or self._websocket.state is not State.OPEN:
             return
 
-        if self._delay_in_frames <= 0:
-            logger.debug("No delay_in_frames set, skipping flush")
-            return
-
-        # Create a silence chunk (zeros) of frame_size samples
-        # Each sample is 2 bytes (16-bit PCM)
-        silence_bytes = bytes(self._frame_size * 2)
-        silence_b64 = base64.b64encode(silence_bytes).decode("utf-8")
-
-        logger.debug(f"Flushing Gradium STT with {self._delay_in_frames} silence frames")
-
-        for _ in range(self._delay_in_frames):
-            msg = {"type": "audio", "audio": silence_b64}
-            try:
-                await self._websocket.send(json.dumps(msg))
-            except Exception as e:
-                logger.warning(f"Failed to send silence frame: {e}")
-                break
+        self._flush_counter += 1
+        flush_id = str(self._flush_counter)
+        msg = {"type": "flush", "flush_id": flush_id}
+        try:
+            await self._websocket.send(json.dumps(msg))
+        except Exception as e:
+            logger.warning(f"Failed to send flush: {e}")
 
     async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
         """Process audio data for speech-to-text conversion.
@@ -353,7 +385,8 @@ class GradiumSTTService(WebsocketSTTService):
             await self._call_event_handler("on_connected")
             setup_msg = {
                 "type": "setup",
-                "input_format": "pcm",
+                "model_name": self._settings.model,
+                "input_format": self._input_format,
             }
             # Build json_config: start with deprecated json_config, then override with params
             json_config = {}
@@ -375,13 +408,7 @@ class GradiumSTTService(WebsocketSTTService):
             if ready_msg["type"] != "ready":
                 raise Exception(f"unexpected first message type {ready_msg['type']}")
 
-            # Store delay_in_frames and frame_size for silence flushing
-            self._delay_in_frames = ready_msg.get("delay_in_frames", 0)
-            self._frame_size = ready_msg.get("frame_size", 1920)
-            logger.debug(
-                f"Connected to Gradium STT (delay_in_frames={self._delay_in_frames}, "
-                f"frame_size={self._frame_size})"
-            )
+            logger.debug("Connected to Gradium STT")
 
         except Exception as e:
             await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
@@ -390,6 +417,13 @@ class GradiumSTTService(WebsocketSTTService):
     async def _disconnect(self):
         await super()._disconnect()
 
+        if self._transcript_aggregation_task:
+            await self.cancel_task(self._transcript_aggregation_task)
+            self._transcript_aggregation_task = None
+
+        self._accumulated_text.clear()
+        self._flush_counter = 0
+
         if self._receive_task:
             await self.cancel_task(self._receive_task)
             self._receive_task = None
@@ -412,41 +446,75 @@ class GradiumSTTService(WebsocketSTTService):
             return self._websocket
         raise Exception("Websocket not connected")
 
-    async def _process_messages(self):
+    async def _receive_messages(self):
         async for message in self._get_websocket():
             try:
-                data = json.loads(message)
-                await self._process_response(data)
+                msg = json.loads(message)
             except json.JSONDecodeError:
                 logger.warning(f"Received non-JSON message: {message}")
+                continue
 
-    async def _receive_messages(self):
-        while True:
-            await self._process_messages()
-            logger.debug(f"{self} Gradium connection was disconnected (timeout?), reconnecting")
-            await self._connect_websocket()
-
-    async def _process_response(self, msg):
-        type_ = msg.get("type", "")
-        if type_ == "text":
-            await self._handle_text(msg["text"])
-        elif type_ == "end_of_stream":
-            await self._handle_end_of_stream()
-        elif type_ == "error":
-            await self.push_error(error_msg=f"Error: {msg}")
-
-    async def _handle_end_of_stream(self):
-        """Handle termination message."""
-        logger.debug("Received end_of_stream message from server")
+            type_ = msg.get("type", "")
+            if type_ == "text":
+                await self._handle_text(msg["text"])
+            elif type_ == "flushed":
+                await self._handle_flushed()
+            elif type_ == "end_of_stream":
+                logger.debug("Received end_of_stream message from server")
+            elif type_ == "error":
+                await self.push_error(error_msg=f"Error: {msg}")
 
     async def _handle_text(self, text: str):
-        """Handle transcription results."""
+        """Handle streaming transcription fragment.
+
+        Accumulates text and pushes an InterimTranscriptionFrame with the
+        full accumulated text so far.
+        """
+        self._accumulated_text.append(text)
+        accumulated = " ".join(self._accumulated_text)
+        await self.push_frame(
+            InterimTranscriptionFrame(
+                text=accumulated,
+                user_id=self._user_id,
+                timestamp=time_now_iso8601(),
+                language=self._settings.language,
+            )
+        )
+        await self.stop_processing_metrics()
+
+    async def _handle_flushed(self):
+        """Handle flush completion by starting a transcript aggregation timer.
+
+        The "flushed" message confirms that buffered audio has been processed,
+        but text tokens may still arrive after this point. A short timer allows
+        trailing tokens to accumulate before finalizing the transcription.
+        """
+        if self._transcript_aggregation_task:
+            await self.cancel_task(self._transcript_aggregation_task)
+        self._transcript_aggregation_task = self.create_task(
+            self._transcript_aggregation_handler(), "transcript_aggregation"
+        )
+
+    async def _transcript_aggregation_handler(self):
+        """Wait for trailing tokens then finalize the accumulated transcription."""
+        await asyncio.sleep(TRANSCRIPT_AGGREGATION_DELAY)
+        await self._finalize_accumulated_text()
+
+    async def _finalize_accumulated_text(self):
+        """Join accumulated text, push TranscriptionFrame, and clear state."""
+        if not self._accumulated_text:
+            return
+        self._transcript_aggregation_task = None
+
+        text = " ".join(self._accumulated_text)
+        self._accumulated_text.clear()
+        logger.debug(f"Final transcription: [{text}]")
         await self.push_frame(
             TranscriptionFrame(
                 text,
                 self._user_id,
                 time_now_iso8601(),
+                self._settings.language,
             )
         )
-        await self._trace_transcription(text, is_final=True, language=None)
-        await self.stop_processing_metrics()
+        await self._trace_transcription(text, is_final=True, language=self._settings.language)

src/pipecat/services/lmnt/tts.py

@@ -21,7 +21,6 @@ from pipecat.frames.frames import (
     TTSAudioRawFrame,
     TTSStoppedFrame,
 )
-from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.settings import TTSSettings
 from pipecat.services.tts_service import InterruptibleTTSService
 from pipecat.transcriptions.language import Language, resolve_language
@@ -212,15 +211,6 @@ class LmntTTSService(InterruptibleTTSService):
         await super().cancel(frame)
         await self._disconnect()
 
-    async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
-        """Push a frame downstream with special handling for stop conditions.
-
-        Args:
-            frame: The frame to push.
-            direction: The direction to push the frame.
-        """
-        await super().push_frame(frame, direction)
-
     async def _connect(self):
         """Connect to LMNT WebSocket and start receive task."""
         await super()._connect()
@@ -322,18 +312,22 @@ class LmntTTSService(InterruptibleTTSService):
             if isinstance(message, bytes):
                 # Raw audio data
                 await self.stop_ttfb_metrics()
+                context_id = self.get_active_audio_context_id()
                 frame = TTSAudioRawFrame(
                     audio=message,
                     sample_rate=self.sample_rate,
                     num_channels=1,
-                    context_id=self.get_active_audio_context_id(),
+                    context_id=context_id,
                 )
-                await self.push_frame(frame)
+                await self.append_to_audio_context(context_id, frame)
             else:
                 try:
                     msg = json.loads(message)
                     if "error" in msg:
-                        await self.push_frame(TTSStoppedFrame())
+                        context_id = self.get_active_audio_context_id()
+                        await self.append_to_audio_context(
+                            context_id, TTSStoppedFrame(context_id=context_id)
+                        )
                         await self.stop_all_metrics()
                         await self.push_error(error_msg=f"Error: {msg['error']}")
                         return

src/pipecat/services/neuphonic/tts.py

@@ -21,18 +21,14 @@ from loguru import logger
 from pydantic import BaseModel
 
 from pipecat.frames.frames import (
-    BotStoppedSpeakingFrame,
     CancelFrame,
     EndFrame,
     ErrorFrame,
     Frame,
-    LLMFullResponseEndFrame,
     StartFrame,
     TTSAudioRawFrame,
-    TTSSpeakFrame,
     TTSStoppedFrame,
 )
-from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.settings import NOT_GIVEN, TTSSettings, _NotGiven
 from pipecat.services.tts_service import InterruptibleTTSService, TextAggregationMode, TTSService
 from pipecat.transcriptions.language import Language, resolve_language
@@ -180,6 +176,7 @@ class NeuphonicTTSService(InterruptibleTTSService):
             text_aggregation_mode=text_aggregation_mode,
             push_stop_frames=True,
             push_start_frame=True,
+            pause_frame_processing=True,
             stop_frame_timeout_s=2.0,
             sample_rate=sample_rate,
             settings=default_settings,
@@ -254,34 +251,6 @@ class NeuphonicTTSService(InterruptibleTTSService):
             msg = {"text": "<STOP>"}
             await self._websocket.send(json.dumps(msg))
 
-    async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
-        """Push a frame downstream with special handling for stop conditions.
-
-        Args:
-            frame: The frame to push.
-            direction: The direction to push the frame.
-        """
-        await super().push_frame(frame, direction)
-
-    async def process_frame(self, frame: Frame, direction: FrameDirection):
-        """Process frames with special handling for speech control.
-
-        Args:
-            frame: The frame to process.
-            direction: The direction of frame processing.
-        """
-        await super().process_frame(frame, direction)
-
-        # If we received a TTSSpeakFrame and the LLM response included text (it
-        # might be that it's only a function calling response) we pause
-        # processing more frames until we receive a BotStoppedSpeakingFrame.
-        if isinstance(frame, TTSSpeakFrame):
-            await self.pause_processing_frames()
-        elif isinstance(frame, LLMFullResponseEndFrame):
-            await self.pause_processing_frames()
-        elif isinstance(frame, BotStoppedSpeakingFrame):
-            await self.resume_processing_frames()
-
     async def _connect(self):
         """Connect to Neuphonic WebSocket and start background tasks."""
         await super()._connect()
@@ -366,10 +335,14 @@ class NeuphonicTTSService(InterruptibleTTSService):
                     await self.stop_ttfb_metrics()
 
                     audio = base64.b64decode(msg["data"]["audio"])
+                    context_id = self.get_active_audio_context_id()
                     frame = TTSAudioRawFrame(
-                        audio, self.sample_rate, 1, context_id=self.get_active_audio_context_id()
+                        audio,
+                        self.sample_rate,
+                        1,
+                        context_id=context_id,
                     )
-                    await self.push_frame(frame)
+                    await self.append_to_audio_context(context_id, frame)
 
     async def _keepalive_task_handler(self):
         """Handle keepalive messages to maintain WebSocket connection."""

src/pipecat/services/openai/__init__.py

@@ -11,6 +11,7 @@ from pipecat.services import DeprecatedModuleProxy
 from .image import *
 from .llm import *
 from .realtime import *
+from .responses.llm import *
 from .stt import *
 from .tts import *
 

src/pipecat/services/openai/responses/__init__.py

@@ -0,0 +1,5 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#

src/pipecat/services/openai/responses/llm.py

@@ -0,0 +1,400 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""OpenAI Responses API LLM service implementation."""
+
+import json
+from contextlib import asynccontextmanager
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Mapping, Optional
+
+import httpx
+from loguru import logger
+from openai import NOT_GIVEN, AsyncOpenAI, AsyncStream, DefaultAsyncHttpxClient
+from openai.types.responses import (
+    ResponseCompletedEvent,
+    ResponseFunctionCallArgumentsDeltaEvent,
+    ResponseFunctionCallArgumentsDoneEvent,
+    ResponseFunctionToolCall,
+    ResponseOutputItemAddedEvent,
+    ResponseOutputItemDoneEvent,
+    ResponseStreamEvent,
+    ResponseTextDeltaEvent,
+)
+
+from pipecat.adapters.services.open_ai_responses_adapter import (
+    OpenAIResponsesLLMAdapter,
+    OpenAIResponsesLLMInvocationParams,
+)
+from pipecat.frames.frames import (
+    Frame,
+    LLMContextFrame,
+    LLMFullResponseEndFrame,
+    LLMFullResponseStartFrame,
+)
+from pipecat.metrics.metrics import LLMTokenUsage
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.frame_processor import FrameDirection
+from pipecat.services.llm_service import FunctionCallFromLLM, LLMService
+from pipecat.services.settings import NOT_GIVEN as _NOT_GIVEN
+from pipecat.services.settings import LLMSettings, _NotGiven
+from pipecat.utils.tracing.service_decorators import traced_llm
+
+
+@dataclass
+class OpenAIResponsesLLMSettings(LLMSettings):
+    """Settings for OpenAIResponsesLLMService.
+
+    Parameters:
+        max_completion_tokens: Maximum completion tokens to generate.
+    """
+
+    max_completion_tokens: int | _NotGiven = field(default_factory=lambda: _NOT_GIVEN)
+
+
+class OpenAIResponsesLLMService(LLMService):
+    """OpenAI Responses API LLM service.
+
+    This service works with the universal LLMContext and LLMContextAggregatorPair.
+
+    Example::
+
+        llm = OpenAIResponsesLLMService(
+            api_key=os.getenv("OPENAI_API_KEY"),
+            settings=OpenAIResponsesLLMService.Settings(
+                model="gpt-4.1",
+                system_instruction="You are a helpful assistant.",
+            ),
+        )
+    """
+
+    Settings = OpenAIResponsesLLMSettings
+    _settings: Settings
+
+    adapter_class = OpenAIResponsesLLMAdapter
+
+    def __init__(
+        self,
+        *,
+        api_key=None,
+        base_url=None,
+        organization=None,
+        project=None,
+        default_headers: Optional[Mapping[str, str]] = None,
+        service_tier: Optional[str] = None,
+        settings: Optional[Settings] = None,
+        **kwargs,
+    ):
+        """Initialize the OpenAI Responses API LLM service.
+
+        Args:
+            api_key: OpenAI API key. If None, uses environment variable.
+            base_url: Custom base URL for OpenAI API. If None, uses default.
+            organization: OpenAI organization ID.
+            project: OpenAI project ID.
+            default_headers: Additional HTTP headers to include in requests.
+            service_tier: Service tier to use (e.g., "auto", "flex", "priority").
+            settings: Runtime-updatable settings.
+            **kwargs: Additional arguments passed to the parent LLMService.
+        """
+        default_settings = self.Settings(
+            model="gpt-4.1",
+            system_instruction=None,
+            frequency_penalty=None,
+            presence_penalty=None,
+            seed=None,
+            temperature=NOT_GIVEN,
+            top_p=NOT_GIVEN,
+            top_k=None,
+            max_tokens=None,
+            max_completion_tokens=NOT_GIVEN,
+            filter_incomplete_user_turns=False,
+            user_turn_completion_config=None,
+            extra={},
+        )
+
+        if settings is not None:
+            default_settings.apply_update(settings)
+
+        super().__init__(
+            settings=default_settings,
+            **kwargs,
+        )
+
+        self._service_tier = service_tier
+        self._client = self._create_client(
+            api_key=api_key,
+            base_url=base_url,
+            organization=organization,
+            project=project,
+            default_headers=default_headers,
+        )
+
+        if self._settings.system_instruction:
+            logger.debug(f"{self}: Using system instruction: {self._settings.system_instruction}")
+
+    def _create_client(
+        self,
+        api_key=None,
+        base_url=None,
+        organization=None,
+        project=None,
+        default_headers=None,
+    ) -> AsyncOpenAI:
+        """Create an AsyncOpenAI client instance.
+
+        Args:
+            api_key: OpenAI API key.
+            base_url: Custom base URL for the API.
+            organization: OpenAI organization ID.
+            project: OpenAI project ID.
+            default_headers: Additional HTTP headers.
+
+        Returns:
+            Configured AsyncOpenAI client instance.
+        """
+        return AsyncOpenAI(
+            api_key=api_key,
+            base_url=base_url,
+            organization=organization,
+            project=project,
+            http_client=DefaultAsyncHttpxClient(
+                limits=httpx.Limits(
+                    max_keepalive_connections=100, max_connections=1000, keepalive_expiry=None
+                )
+            ),
+            default_headers=default_headers,
+        )
+
+    def can_generate_metrics(self) -> bool:
+        """Check if this service can generate processing metrics."""
+        return True
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        """Process frames for LLM completion requests.
+
+        Args:
+            frame: The frame to process.
+            direction: The direction of frame processing.
+        """
+        await super().process_frame(frame, direction)
+
+        context = None
+        if isinstance(frame, LLMContextFrame):
+            context = frame.context
+        else:
+            await self.push_frame(frame, direction)
+
+        if context:
+            try:
+                await self.push_frame(LLMFullResponseStartFrame())
+                await self.start_processing_metrics()
+                await self._process_context(context)
+            except httpx.TimeoutException as e:
+                await self._call_event_handler("on_completion_timeout")
+                await self.push_error(error_msg="LLM completion timeout", exception=e)
+            except Exception as e:
+                await self.push_error(error_msg=f"Error during completion: {e}", exception=e)
+            finally:
+                await self.stop_processing_metrics()
+                await self.push_frame(LLMFullResponseEndFrame())
+
+    @traced_llm
+    async def _process_context(self, context: LLMContext):
+        adapter: OpenAIResponsesLLMAdapter = self.get_llm_adapter()
+        logger.debug(
+            f"{self}: Generating response from universal context "
+            f"{adapter.get_messages_for_logging(context)}"
+        )
+
+        invocation_params = adapter.get_llm_invocation_params(
+            context, system_instruction=self._settings.system_instruction
+        )
+
+        params = self._build_response_params(invocation_params)
+
+        await self.start_ttfb_metrics()
+
+        stream: AsyncStream[ResponseStreamEvent] = await self._client.responses.create(**params)
+
+        # Track function calls across stream events
+        function_calls: Dict[str, Dict[str, str]] = {}  # item_id -> {name, call_id, arguments}
+        current_arguments: Dict[str, str] = {}  # item_id -> accumulated arguments
+
+        # Ensure stream and its async iterator are closed on cancellation/exception
+        # to prevent socket leaks and uvloop crashes. Closing the iterator first
+        # cascades cleanup through nested async generators (httpx/httpcore internals),
+        # preventing uvloop's broken asyncgen finalizer from firing on Python 3.12+
+        # (MagicStack/uvloop#699).
+        @asynccontextmanager
+        async def _closing(stream):
+            chunk_iter = stream.__aiter__()
+            try:
+                yield chunk_iter
+            finally:
+                # Close the iterator first to cascade cleanup through
+                # nested async generators (httpx/httpcore internals).
+                if hasattr(chunk_iter, "aclose"):
+                    await chunk_iter.aclose()
+                # Then close the stream to release HTTP resources.
+                if hasattr(stream, "close"):
+                    await stream.close()
+                elif hasattr(stream, "aclose"):
+                    await stream.aclose()
+
+        async with _closing(stream) as event_iter:
+            async for event in event_iter:
+                if isinstance(event, ResponseTextDeltaEvent):
+                    await self.stop_ttfb_metrics()
+                    await self._push_llm_text(event.delta)
+
+                elif isinstance(event, ResponseOutputItemAddedEvent):
+                    await self.stop_ttfb_metrics()
+                    item = event.item
+                    if isinstance(item, ResponseFunctionToolCall):
+                        item_id = item.id or ""
+                        function_calls[item_id] = {
+                            "name": item.name,
+                            "call_id": item.call_id,
+                            "arguments": "",
+                        }
+                        current_arguments[item_id] = ""
+
+                elif isinstance(event, ResponseFunctionCallArgumentsDeltaEvent):
+                    item_id = event.item_id
+                    if item_id in current_arguments:
+                        current_arguments[item_id] += event.delta
+
+                elif isinstance(event, ResponseFunctionCallArgumentsDoneEvent):
+                    item_id = event.item_id
+                    if item_id in function_calls:
+                        function_calls[item_id]["arguments"] = event.arguments
+
+                elif isinstance(event, ResponseOutputItemDoneEvent):
+                    item = event.item
+                    if isinstance(item, ResponseFunctionToolCall):
+                        item_id = item.id or ""
+                        if item_id in function_calls:
+                            function_calls[item_id]["name"] = item.name
+                            function_calls[item_id]["call_id"] = item.call_id
+                            function_calls[item_id]["arguments"] = item.arguments
+
+                elif isinstance(event, ResponseCompletedEvent):
+                    response = event.response
+                    if response.usage:
+                        tokens = LLMTokenUsage(
+                            prompt_tokens=response.usage.input_tokens,
+                            completion_tokens=response.usage.output_tokens,
+                            total_tokens=response.usage.total_tokens,
+                            cache_read_input_tokens=response.usage.input_tokens_details.cached_tokens,
+                            reasoning_tokens=response.usage.output_tokens_details.reasoning_tokens,
+                        )
+                        await self.start_llm_usage_metrics(tokens)
+
+                    # This field is used by @traced_llm for more detailed
+                    # model name in tracing spans
+                    self._full_model_name = response.model
+
+        # Process any function calls
+        if function_calls:
+            fc_list: List[FunctionCallFromLLM] = []
+            for item_id, fc in function_calls.items():
+                try:
+                    arguments = json.loads(fc["arguments"]) if fc["arguments"] else {}
+                except json.JSONDecodeError:
+                    logger.warning(
+                        f"{self}: Failed to parse function call arguments: {fc['arguments']}"
+                    )
+                    arguments = {}
+                fc_list.append(
+                    FunctionCallFromLLM(
+                        context=context,
+                        tool_call_id=fc["call_id"],
+                        function_name=fc["name"],
+                        arguments=arguments,
+                    )
+                )
+            await self.run_function_calls(fc_list)
+
+    def _build_response_params(self, invocation_params: OpenAIResponsesLLMInvocationParams) -> dict:
+        """Build parameters for the responses.create() call.
+
+        Args:
+            invocation_params: Parameters derived from the LLM context.
+
+        Returns:
+            Dictionary of parameters for the Responses API call.
+        """
+        params: Dict[str, Any] = {
+            "model": self._settings.model,
+            "stream": True,
+            "store": False,
+            "input": invocation_params["input"],
+        }
+
+        # instructions (set by the adapter when input is non-empty)
+        if "instructions" in invocation_params:
+            params["instructions"] = invocation_params["instructions"]
+
+        # Optional parameters - only include if given
+        if isinstance(self._settings.temperature, (int, float)):
+            params["temperature"] = self._settings.temperature
+
+        if isinstance(self._settings.top_p, (int, float)):
+            params["top_p"] = self._settings.top_p
+
+        if isinstance(self._settings.max_completion_tokens, int):
+            params["max_output_tokens"] = self._settings.max_completion_tokens
+
+        if self._service_tier is not None:
+            params["service_tier"] = self._service_tier
+
+        # Tools
+        tools = invocation_params.get("tools")
+        if tools is not None and not isinstance(tools, type(NOT_GIVEN)):
+            params["tools"] = tools
+
+        # Extra settings
+        params.update(self._settings.extra)
+
+        return params
+
+    async def run_inference(
+        self,
+        context: LLMContext,
+        max_tokens: Optional[int] = None,
+        system_instruction: Optional[str] = None,
+    ) -> Optional[str]:
+        """Run a one-shot, out-of-band inference with the given LLM context.
+
+        Args:
+            context: The LLM context containing conversation history.
+            max_tokens: Optional maximum number of tokens to generate.
+            system_instruction: Optional system instruction for this inference.
+
+        Returns:
+            The LLM's response as a string, or None if no response is generated.
+        """
+        adapter: OpenAIResponsesLLMAdapter = self.get_llm_adapter()
+        effective_instruction = system_instruction or self._settings.system_instruction
+        invocation_params = adapter.get_llm_invocation_params(
+            context, system_instruction=effective_instruction
+        )
+
+        params = self._build_response_params(invocation_params)
+
+        # Override for non-streaming
+        params["stream"] = False
+
+        if max_tokens is not None:
+            params["max_output_tokens"] = max_tokens
+
+        response = await self._client.responses.create(**params)
+
+        return response.output_text
+
+
+__all__ = ["OpenAIResponsesLLMService", "OpenAIResponsesLLMSettings"]

src/pipecat/services/openai/stt.py

@@ -358,8 +358,8 @@ class OpenAIRealtimeSTTService(WebsocketSTTService):
         Returns:
             Two-letter ISO-639-1 language code.
         """
-        # Language.value is e.g. "en", "en-US", "fr", "zh".
-        return language.value.split("-")[0].lower()
+        # Language value is e.g. "en", "en-US", "fr", "zh".
+        return str(language).split("-")[0].lower()
 
     def can_generate_metrics(self) -> bool:
         """Check if the service can generate processing metrics.

src/pipecat/services/rime/tts.py

@@ -1054,15 +1054,6 @@ class RimeNonJsonTTSService(InterruptibleTTSService):
         await super().cancel(frame)
         await self._disconnect()
 
-    async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
-        """Push a frame downstream with special handling for stop conditions.
-
-        Args:
-            frame: The frame to push.
-            direction: The direction to push the frame.
-        """
-        await super().push_frame(frame, direction)
-
     async def _connect(self):
         """Establish WebSocket connection and start receive task."""
         await super()._connect()
@@ -1153,13 +1144,14 @@ class RimeNonJsonTTSService(InterruptibleTTSService):
                 if isinstance(message, bytes):
                     await self.stop_ttfb_metrics()
 
+                    context_id = self.get_active_audio_context_id()
                     frame = TTSAudioRawFrame(
                         audio=message,
                         sample_rate=self.sample_rate,
                         num_channels=1,
-                        context_id=self.get_active_audio_context_id(),
+                        context_id=context_id,
                     )
-                    await self.push_frame(frame)
+                    await self.append_to_audio_context(context_id, frame)
             except Exception as e:
                 await self.push_error(error_msg=f"Error: {e}", exception=e)
 

src/pipecat/services/sarvam/tts.py

@@ -1031,23 +1031,6 @@ class SarvamTTSService(InterruptibleTTSService):
         except Exception as e:
             await self.push_error(error_msg=f"Error sending flush to Sarvam: {e}", exception=e)
 
-    async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
-        """Push a frame downstream with special handling for stop conditions.
-
-        Args:
-            frame: The frame to push.
-            direction: The direction to push the frame.
-        """
-        await super().push_frame(frame, direction)
-
-    async def process_frame(self, frame: Frame, direction: FrameDirection):
-        """Process a frame and flush audio if it's the end of a full response."""
-        await super().process_frame(frame, direction)
-
-        # When the LLM finishes responding, flush any remaining text in Sarvam's buffer
-        if isinstance(frame, (LLMFullResponseEndFrame, EndFrame)):
-            await self.flush_audio()
-
     async def _update_settings(self, delta: TTSSettings) -> dict[str, Any]:
         """Apply a settings delta and resend config if voice changed."""
         changed = await super()._update_settings(delta)
@@ -1168,14 +1151,13 @@ class SarvamTTSService(InterruptibleTTSService):
         async for message in self._get_websocket():
             if isinstance(message, str):
                 msg = json.loads(message)
+                context_id = self.get_active_audio_context_id()
                 if msg.get("type") == "audio":
                     # Check for interruption before processing audio
                     await self.stop_ttfb_metrics()
                     audio = base64.b64decode(msg["data"]["audio"])
-                    frame = TTSAudioRawFrame(
-                        audio, self.sample_rate, 1, context_id=self.get_active_audio_context_id()
-                    )
-                    await self.push_frame(frame)
+                    frame = TTSAudioRawFrame(audio, self.sample_rate, 1, context_id=context_id)
+                    await self.append_to_audio_context(context_id, frame)
                 elif msg.get("type") == "error":
                     error_msg = msg["data"]["message"]
                     await self.push_error(error_msg=f"TTS Error: {error_msg}")
@@ -1183,8 +1165,9 @@ class SarvamTTSService(InterruptibleTTSService):
                     # If it's a timeout error, the connection might need to be reset
                     if "too long" in error_msg.lower() or "timeout" in error_msg.lower():
                         logger.warning("Connection timeout detected, service may need restart")
-
-                    await self.push_frame(ErrorFrame(error=f"TTS Error: {error_msg}"))
+                    await self.append_to_audio_context(
+                        context_id, ErrorFrame(error=f"TTS Error: {error_msg}")
+                    )
 
     async def _keepalive_task_handler(self):
         """Handle keepalive messages to maintain WebSocket connection."""

src/pipecat/services/soniox/stt.py

@@ -27,7 +27,7 @@ from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.settings import NOT_GIVEN, STTSettings, _NotGiven
 from pipecat.services.stt_latency import SONIOX_TTFS_P99
 from pipecat.services.stt_service import WebsocketSTTService
-from pipecat.transcriptions.language import Language
+from pipecat.transcriptions.language import Language, resolve_language
 from pipecat.utils.time import time_now_iso8601
 from pipecat.utils.tracing.service_decorators import traced_stt
 
@@ -118,14 +118,75 @@ def is_end_token(token: dict) -> bool:
 
 
 def language_to_soniox_language(language: Language) -> str:
-    """Pipecat Language enum uses same ISO 2-letter codes as Soniox, except with added regional variants.
+    """Convert a Pipecat Language to a Soniox language code.
 
-    For a list of all supported languages, see: https://soniox.com/docs/speech-to-text/core-concepts/supported-languages
+    For a list of all supported languages, see:
+    https://soniox.com/docs/speech-to-text/core-concepts/supported-languages
     """
-    lang_str = str(language.value).lower()
-    if "-" in lang_str:
-        return lang_str.split("-")[0]
-    return lang_str
+    LANGUAGE_MAP = {
+        Language.AF: "af",
+        Language.AR: "ar",
+        Language.AZ: "az",
+        Language.BE: "be",
+        Language.BG: "bg",
+        Language.BN: "bn",
+        Language.BS: "bs",
+        Language.CA: "ca",
+        Language.CS: "cs",
+        Language.CY: "cy",
+        Language.DA: "da",
+        Language.DE: "de",
+        Language.EL: "el",
+        Language.EN: "en",
+        Language.ES: "es",
+        Language.ET: "et",
+        Language.EU: "eu",
+        Language.FA: "fa",
+        Language.FI: "fi",
+        Language.FR: "fr",
+        Language.GL: "gl",
+        Language.GU: "gu",
+        Language.HE: "he",
+        Language.HI: "hi",
+        Language.HR: "hr",
+        Language.HU: "hu",
+        Language.ID: "id",
+        Language.IT: "it",
+        Language.JA: "ja",
+        Language.KA: "ka",
+        Language.KK: "kk",
+        Language.KN: "kn",
+        Language.KO: "ko",
+        Language.LT: "lt",
+        Language.LV: "lv",
+        Language.MK: "mk",
+        Language.ML: "ml",
+        Language.MR: "mr",
+        Language.MS: "ms",
+        Language.NL: "nl",
+        Language.NO: "no",
+        Language.PA: "pa",
+        Language.PL: "pl",
+        Language.PT: "pt",
+        Language.RO: "ro",
+        Language.RU: "ru",
+        Language.SK: "sk",
+        Language.SL: "sl",
+        Language.SQ: "sq",
+        Language.SR: "sr",
+        Language.SV: "sv",
+        Language.SW: "sw",
+        Language.TA: "ta",
+        Language.TE: "te",
+        Language.TH: "th",
+        Language.TL: "tl",
+        Language.TR: "tr",
+        Language.UK: "uk",
+        Language.UR: "ur",
+        Language.VI: "vi",
+        Language.ZH: "zh",
+    }
+    return resolve_language(language, LANGUAGE_MAP, use_base_code=True)
 
 
 def _prepare_language_hints(

src/pipecat/services/stt_latency.py

@@ -46,6 +46,7 @@ SAMBANOVA_TTFS_P99: float = 2.20
 SARVAM_TTFS_P99: float = 1.17
 SONIOX_TTFS_P99: float = 0.35
 SPEECHMATICS_TTFS_P99: float = 0.74
+TOGETHER_TTFS_P99: float = 2.028
 
 # These services run locally and should be replaced with measured values
 NVIDIA_TTFS_P99: float = DEFAULT_TTFS_P99

src/pipecat/services/stt_service.py

@@ -124,6 +124,18 @@ class STTService(AIService):
         # Convert Language enum to service-specific format at init time.
         # Runtime updates are handled by _update_settings(), but init-time
         # settings bypass that path and need explicit conversion.
+        # Raw strings (e.g. "de-DE") are first converted to Language enums
+        # so they go through the same resolution logic.
+        if isinstance(self._settings.language, str) and not isinstance(
+            self._settings.language, Language
+        ):
+            try:
+                self._settings.language = Language(self._settings.language)
+            except ValueError:
+                logger.warning(
+                    f"Language string '{self._settings.language}' is not a recognized "
+                    f"Language code. It will be passed to the service as-is."
+                )
         if isinstance(self._settings.language, Language):
             converted = self.language_to_service_language(self._settings.language)
             if converted is not None:
@@ -294,7 +306,20 @@ class STTService(AIService):
         Returns:
             Dict mapping changed field names to their previous values.
         """
-        # Translate language *before* applying so the stored value is canonical
+        # Translate language *before* applying so the stored value is canonical.
+        # Raw strings are first converted to Language enums for proper resolution.
+        if (
+            is_given(delta.language)
+            and isinstance(delta.language, str)
+            and not isinstance(delta.language, Language)
+        ):
+            try:
+                delta.language = Language(delta.language)
+            except ValueError:
+                logger.warning(
+                    f"Language string '{delta.language}' is not a recognized "
+                    f"Language code. It will be passed to the service as-is."
+                )
         if is_given(delta.language) and isinstance(delta.language, Language):
             converted = self.language_to_service_language(delta.language)
             if converted is not None:

src/pipecat/services/together/__init__.py

@@ -9,5 +9,7 @@ import sys
 from pipecat.services import DeprecatedModuleProxy
 
 from .llm import *
+from .stt import *
+from .tts import *
 
 sys.modules[__name__] = DeprecatedModuleProxy(globals(), "together", "together.llm")

src/pipecat/services/together/llm.py

@@ -36,7 +36,7 @@ class TogetherLLMService(OpenAILLMService):
         self,
         *,
         api_key: str,
-        base_url: str = "https://api.together.xyz/v1",
+        base_url: str = "https://api.together.ai/v1",
         model: Optional[str] = None,
         settings: Optional[Settings] = None,
         **kwargs,
@@ -45,8 +45,8 @@ class TogetherLLMService(OpenAILLMService):
 
         Args:
             api_key: The API key for accessing Together.ai's API.
-            base_url: The base URL for Together.ai API. Defaults to "https://api.together.xyz/v1".
-            model: The model identifier to use. Defaults to "meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo".
+            base_url: The base URL for Together.ai API. Defaults to "https://api.together.ai/v1".
+            model: The model identifier to use.
 
                 .. deprecated:: 0.0.105
                     Use ``settings=TogetherLLMService.Settings(model=...)`` instead.
@@ -56,7 +56,9 @@ class TogetherLLMService(OpenAILLMService):
             **kwargs: Additional keyword arguments passed to OpenAILLMService.
         """
         # 1. Initialize default_settings with hardcoded defaults
-        default_settings = self.Settings(model="meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo")
+        default_settings = self.Settings(
+            model="openai/gpt-oss-120b",
+        )
 
         # 2. Apply direct init arg overrides (deprecated)
         if model is not None:

src/pipecat/services/together/stt.py

@@ -0,0 +1,452 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Together AI speech-to-text service implementation."""
+
+import base64
+import json
+from dataclasses import dataclass
+from typing import Any, AsyncGenerator, Optional
+
+from loguru import logger
+
+from pipecat.audio.utils import create_stream_resampler
+from pipecat.services.settings import STTSettings
+from pipecat.services.stt_latency import TOGETHER_TTFS_P99
+
+try:
+    from websockets.asyncio.client import connect as websocket_connect
+    from websockets.protocol import State
+except ModuleNotFoundError as e:
+    logger.error(f"Exception: {e}")
+    logger.error("In order to use Together, you need to `pip install pipecat-ai[together]`.")
+    raise Exception(f"Missing module: {e}")
+
+from pipecat.frames.frames import (
+    CancelFrame,
+    EndFrame,
+    Frame,
+    InterimTranscriptionFrame,
+    StartFrame,
+    TranscriptionFrame,
+    VADUserStartedSpeakingFrame,
+    VADUserStoppedSpeakingFrame,
+)
+from pipecat.processors.frame_processor import FrameDirection
+from pipecat.services.stt_service import WebsocketSTTService
+from pipecat.transcriptions.language import Language
+from pipecat.utils.time import time_now_iso8601
+from pipecat.utils.tracing.service_decorators import traced_stt
+
+# Together requires 16 kHz 16-bit mono PCM input.
+_TOGETHER_SAMPLE_RATE = 16000
+
+
+@dataclass
+class TogetherSTTSettings(STTSettings):
+    """Settings for the Together AI STT service.
+
+    Parameters:
+        model: Together AI transcription model to use.
+        language: Language of the audio input.
+    """
+
+    pass
+
+
+class TogetherSTTService(WebsocketSTTService):
+    """Together AI speech-to-text service.
+
+    Provides real-time speech recognition using Together AI's WebSocket API
+    with OpenAI-compatible speech-to-text endpoints.
+
+    Example::
+
+        stt = TogetherSTTService(
+            api_key="...",
+            settings=TogetherSTTService.Settings(
+                model="openai/whisper-large-v3",
+            ),
+        )
+    """
+
+    Settings = TogetherSTTSettings
+    _settings: Settings
+
+    def __init__(
+        self,
+        *,
+        api_key: str,
+        base_url: str = "wss://api.together.ai/v1",
+        settings: Optional[Settings] = None,
+        ttfs_p99_latency: float = TOGETHER_TTFS_P99,
+        **kwargs,
+    ):
+        """Initialize the Together AI STT service.
+
+        Args:
+            api_key: Together AI API key for authentication.
+            base_url: The URL of the Together AI WebSocket API.
+            settings: Runtime-updatable settings for model and language configuration.
+            ttfs_p99_latency: P99 latency from speech end to final transcript in seconds.
+                Override for your deployment. See https://github.com/pipecat-ai/stt-benchmark
+            **kwargs: Additional arguments passed to the parent WebsocketSTTService.
+        """
+        # Hardcoded defaults
+        default_settings = self.Settings(
+            model="openai/whisper-large-v3",
+            language=Language.EN,
+        )
+
+        # Apply settings delta
+        if settings is not None:
+            default_settings.apply_update(settings)
+
+        super().__init__(
+            ttfs_p99_latency=ttfs_p99_latency,
+            keepalive_timeout=20,
+            keepalive_interval=5,
+            settings=default_settings,
+            **kwargs,
+        )
+
+        self._api_key = api_key
+        self._base_url = base_url
+        self._receive_task = None
+        self._resampler = create_stream_resampler()
+
+    def can_generate_metrics(self) -> bool:
+        """Check if this service can generate processing metrics.
+
+        Returns:
+            True, as Together STT service supports metrics generation.
+        """
+        return True
+
+    async def _update_settings(self, delta: STTSettings) -> dict[str, Any]:
+        """Apply a settings delta and reconnect to apply changes.
+
+        Together passes model/language as URL query params, so a reconnect
+        is needed to apply changes.
+
+        Args:
+            delta: A settings delta with updated values.
+
+        Returns:
+            Dict mapping changed field names to their previous values.
+        """
+        changed = await super()._update_settings(delta)
+
+        if not changed:
+            return changed
+
+        # Reconnect to apply updated settings (they become WS URL params)
+        await self._disconnect()
+        await self._connect()
+
+        return changed
+
+    async def _send_keepalive(self, silence: bytes):
+        """Send silent audio to keep the Together AI connection alive.
+
+        Wraps silence in the ``input_audio_buffer.append`` JSON protocol.
+
+        Args:
+            silence: Silent 16-bit mono PCM audio bytes.
+        """
+        await self._send_audio(silence)
+
+    async def start(self, frame: StartFrame):
+        """Start the Together AI STT service.
+
+        Args:
+            frame: The start frame containing initialization parameters.
+        """
+        await super().start(frame)
+        await self._connect()
+
+    async def stop(self, frame: EndFrame):
+        """Stop the Together AI STT service.
+
+        Args:
+            frame: The end frame.
+        """
+        await super().stop(frame)
+        await self._disconnect()
+
+    async def cancel(self, frame: CancelFrame):
+        """Cancel the Together AI STT service.
+
+        Args:
+            frame: The cancel frame.
+        """
+        await super().cancel(frame)
+        await self._disconnect()
+
+    async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
+        """Send audio data to Together AI for transcription.
+
+        Args:
+            audio: Raw audio bytes to transcribe.
+
+        Yields:
+            Frame: None (transcription results come via WebSocket callbacks).
+        """
+        await self._send_audio(audio)
+        yield None
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        """Process frames with Together AI-specific handling.
+
+        Args:
+            frame: The frame to process.
+            direction: The direction of frame flow in the pipeline.
+        """
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, VADUserStartedSpeakingFrame):
+            await self.start_processing_metrics()
+        elif isinstance(frame, VADUserStoppedSpeakingFrame):
+            await self._commit_audio_buffer()
+
+    # ------------------------------------------------------------------
+    # WebSocket connection management
+    # ------------------------------------------------------------------
+
+    async def _connect(self):
+        """Connect to the transcription endpoint and start receiving."""
+        await super()._connect()
+        await self._connect_websocket()
+        if self._websocket and not self._receive_task:
+            self._receive_task = self.create_task(self._receive_task_handler(self._report_error))
+
+    async def _disconnect(self):
+        """Disconnect and clean up background tasks."""
+        await super()._disconnect()
+        if self._receive_task:
+            await self.cancel_task(self._receive_task, timeout=1.0)
+            self._receive_task = None
+        await self._disconnect_websocket()
+
+    async def _connect_websocket(self):
+        """Establish the WebSocket connection to the Together AI endpoint."""
+        try:
+            if self._websocket and self._websocket.state is State.OPEN:
+                return
+
+            url = (
+                f"{self._base_url}/realtime?intent=transcription"
+                f"&model={self._settings.model}"
+                f"&input_audio_format=pcm_s16le_16000"
+            )
+            headers = {
+                "Authorization": f"Bearer {self._api_key}",
+            }
+
+            self._websocket = await websocket_connect(url, additional_headers=headers)
+        except Exception as e:
+            await self.push_error(
+                error_msg=f"Error connecting to Together AI STT: {e}",
+                exception=e,
+            )
+            self._websocket = None
+
+    async def _disconnect_websocket(self):
+        """Close the WebSocket connection."""
+        try:
+            if self._websocket:
+                await self._websocket.close()
+        except Exception as e:
+            await self.push_error(
+                error_msg=f"Error disconnecting: {e}",
+                exception=e,
+            )
+        finally:
+            self._websocket = None
+            await self._call_event_handler("on_disconnected")
+
+    # ------------------------------------------------------------------
+    # Client events
+    # ------------------------------------------------------------------
+
+    async def _send_audio(self, audio: bytes):
+        """Send audio data via ``input_audio_buffer.append``.
+
+        Resamples from the pipeline sample rate to 16 kHz if needed.
+
+        Args:
+            audio: Raw audio bytes at the pipeline sample rate.
+        """
+        try:
+            if not self._disconnecting and self._websocket:
+                audio = await self._resampler.resample(
+                    audio, self.sample_rate, _TOGETHER_SAMPLE_RATE
+                )
+                if not audio:
+                    return
+                payload = base64.b64encode(audio).decode("utf-8")
+                await self._websocket.send(
+                    json.dumps({"type": "input_audio_buffer.append", "audio": payload})
+                )
+        except Exception as e:
+            if self._disconnecting or not self._websocket:
+                return
+            await self.push_error(
+                error_msg=f"Error sending audio: {e}",
+                exception=e,
+            )
+
+    async def _commit_audio_buffer(self):
+        """Commit the current audio buffer for transcription."""
+        try:
+            if not self._disconnecting and self._websocket:
+                await self._websocket.send(json.dumps({"type": "input_audio_buffer.commit"}))
+        except Exception as e:
+            if self._disconnecting or not self._websocket:
+                return
+            await self.push_error(
+                error_msg=f"Error committing audio buffer: {e}",
+                exception=e,
+            )
+
+    # ------------------------------------------------------------------
+    # Server event handling
+    # ------------------------------------------------------------------
+
+    async def _receive_messages(self):
+        """Receive and dispatch server events from the transcription session.
+
+        Called by ``WebsocketService._receive_task_handler`` which wraps
+        this method with automatic reconnection on connection errors.
+        """
+        async for message in self._websocket:
+            try:
+                evt = json.loads(message)
+            except json.JSONDecodeError:
+                logger.warning(f"{self} failed to parse WebSocket message")
+                continue
+
+            evt_type = evt.get("type", "")
+
+            if evt_type == "session.created":
+                await self._handle_session_created(evt)
+            elif evt_type == "conversation.item.input_audio_transcription.delta":
+                await self._handle_transcription_delta(evt)
+            elif evt_type == "conversation.item.input_audio_transcription.completed":
+                await self._handle_transcription_completed(evt)
+            elif evt_type == "conversation.item.input_audio_transcription.failed":
+                await self._handle_transcription_failed(evt)
+            elif evt_type == "input_audio_buffer.committed":
+                logger.trace(f"Audio buffer committed: item_id={evt.get('item_id', '')}")
+            elif evt_type == "error":
+                await self._handle_error(evt)
+            else:
+                logger.trace(f"{self} unhandled event: {evt_type}")
+
+    async def _handle_session_created(self, evt: dict):
+        """Handle ``session.created`` event.
+
+        Args:
+            evt: The session created event from the server.
+        """
+        session = evt.get("session", {})
+        self._session_id = session.get("id")
+        logger.debug(f"{self} session created: {self._session_id}")
+
+        upgrade_session_message = {
+            "type": "transcription_session.update",
+            "session": {
+                "input_audio_format": "pcm_s16le_16000",
+                "input_audio_transcription": {
+                    "model": self._settings.model,
+                    "language": self._settings.language,
+                    "prompt": "Transcribe the incoming audio in real time.",
+                },
+                "turn_detection": {
+                    "type": "none",
+                },
+            },
+        }
+
+        await self._websocket.send(json.dumps(upgrade_session_message))
+        await self._call_event_handler("on_connected")
+
+    async def _handle_transcription_delta(self, evt: dict):
+        """Handle incremental transcription text.
+
+        Args:
+            evt: The delta event from the server.
+        """
+        delta = evt.get("delta", "")
+        if delta.strip():
+            await self.push_frame(
+                InterimTranscriptionFrame(
+                    delta,
+                    self._user_id,
+                    time_now_iso8601(),
+                    self._settings.language,
+                    result=evt,
+                )
+            )
+
+    async def _handle_transcription_completed(self, evt: dict):
+        """Handle a completed transcription for a speech segment.
+
+        Args:
+            evt: The completed event containing the full transcript.
+        """
+        transcript = evt.get("transcript", "").strip()
+        if transcript:
+            await self.push_frame(
+                TranscriptionFrame(
+                    transcript,
+                    self._user_id,
+                    time_now_iso8601(),
+                    self._settings.language,
+                    result=evt,
+                    finalized=True,
+                )
+            )
+            await self._handle_transcription_trace(transcript, True, self._settings.language)
+            await self.stop_processing_metrics()
+
+    @traced_stt
+    async def _handle_transcription_trace(
+        self, transcript: str, is_final: bool, language: Optional[Language] = None
+    ):
+        """Record transcription result for tracing.
+
+        Args:
+            transcript: The transcribed text.
+            is_final: Whether this is a final transcription result.
+            language: Optional language of the transcription.
+        """
+        pass
+
+    async def _handle_transcription_failed(self, evt: dict):
+        """Handle a transcription failure for a speech segment.
+
+        Args:
+            evt: The failed event containing error details.
+        """
+        error_info = evt.get("error", {})
+        await self.push_error(error_msg=f"Transcription failed: {error_info}")
+
+    async def _handle_error(self, evt: dict):
+        """Handle a fatal error from the transcription session.
+
+        Raises an exception so that ``WebsocketService`` can decide
+        whether to attempt reconnection.
+
+        Args:
+            evt: The error event.
+        """
+        error_info = evt.get("error", {})
+        error_msg = error_info.get("message", "Unknown error")
+        error_code = error_info.get("code", "")
+        msg = f"Together AI STT error [{error_code}]: {error_msg}"
+        await self.push_error(error_msg=msg)
+        raise Exception(msg)

src/pipecat/services/together/tts.py

@@ -0,0 +1,468 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Together AI text-to-speech service implementation."""
+
+import base64
+import json
+from dataclasses import dataclass
+from typing import Any, AsyncGenerator, Optional
+
+from loguru import logger
+
+from pipecat.services.settings import TTSSettings
+
+try:
+    from websockets.asyncio.client import connect as websocket_connect
+    from websockets.protocol import State
+except ModuleNotFoundError as e:
+    logger.error(f"Exception: {e}")
+    logger.error("In order to use Together, you need to `pip install pipecat-ai[together]`.")
+    raise Exception(f"Missing module: {e}")
+
+from pipecat.frames.frames import (
+    CancelFrame,
+    EndFrame,
+    Frame,
+    StartFrame,
+    TTSAudioRawFrame,
+    TTSStoppedFrame,
+)
+from pipecat.services.tts_service import WebsocketTTSService
+from pipecat.transcriptions.language import Language
+from pipecat.utils.tracing.service_decorators import traced_tts
+
+
+@dataclass
+class TogetherTTSSettings(TTSSettings):
+    """Settings for the Together AI TTS service.
+
+    Parameters:
+        max_partial_length: Maximum partial text length for streaming.
+    """
+
+    max_partial_length: Optional[int] = None
+
+
+class TogetherTTSService(WebsocketTTSService):
+    """Together AI TTS service with WebSocket streaming.
+
+    Provides text-to-speech using Together AI's realtime WebSocket API.
+    Supports streaming synthesis with configurable voice and model options.
+    """
+
+    Settings = TogetherTTSSettings
+    _settings: Settings
+
+    def __init__(
+        self,
+        *,
+        api_key: str,
+        url: str = "wss://api.together.ai/v1/audio/speech/websocket",
+        sample_rate: Optional[int] = None,
+        settings: Optional[Settings] = None,
+        **kwargs,
+    ):
+        """Initialize the Together AI TTS service.
+
+        Args:
+            api_key: Together AI API key for authentication.
+            url: WebSocket URL for Together AI TTS API.
+            sample_rate: Audio sample rate (default: 24000).
+            settings: Runtime-updatable settings for model, voice, and language configuration.
+            **kwargs: Additional arguments passed to the parent service.
+        """
+        # Hardcoded defaults
+        default_settings = self.Settings(
+            model="canopylabs/orpheus-3b-0.1-ft",
+            voice="tara",
+            language=Language.EN,
+        )
+
+        # Apply settings delta
+        if settings is not None:
+            default_settings.apply_update(settings)
+
+        super().__init__(
+            sample_rate=sample_rate,
+            push_start_frame=True,
+            settings=default_settings,
+            **kwargs,
+        )
+
+        self._api_key = api_key
+        self._url = url
+        self._session_id = None
+        self._receive_task = None
+        self._context_id: Optional[str] = None
+        self._pending_commits = 0
+        self._flush_context_id: Optional[str] = None
+
+    def can_generate_metrics(self) -> bool:
+        """Check if this service can generate processing metrics.
+
+        Returns:
+            True, as Together TTS service supports metrics generation.
+        """
+        return True
+
+    async def _update_settings(self, delta: TTSSettings) -> dict[str, Any]:
+        """Apply a settings delta and reconnect to apply changes.
+
+        Together passes model/voice as URL query params, so a reconnect
+        is needed to apply changes.
+
+        Args:
+            delta: A settings delta with updated values.
+
+        Returns:
+            Dict mapping changed field names to their previous values.
+        """
+        changed = await super()._update_settings(delta)
+
+        if not changed:
+            return changed
+
+        # Reconnect to apply updated settings (they become WS URL params)
+        await self._disconnect()
+        await self._connect()
+
+        return changed
+
+    def _build_websocket_url(self) -> str:
+        """Build the WebSocket URL with query parameters."""
+        url = f"{self._url}?model={self._settings.model}&voice={self._settings.voice}"
+        if self._settings.max_partial_length is not None:
+            url += f"&max_partial_length={self._settings.max_partial_length}"
+        return url
+
+    async def start(self, frame: StartFrame):
+        """Start the Together AI TTS service.
+
+        Args:
+            frame: The start frame containing initialization parameters.
+        """
+        await super().start(frame)
+        await self._connect()
+
+    async def stop(self, frame: EndFrame):
+        """Stop the Together AI TTS service.
+
+        Args:
+            frame: The end frame.
+        """
+        await super().stop(frame)
+        await self._disconnect()
+
+    async def cancel(self, frame: CancelFrame):
+        """Cancel the Together AI TTS service.
+
+        Args:
+            frame: The cancel frame.
+        """
+        await super().cancel(frame)
+        await self._disconnect()
+
+    # ------------------------------------------------------------------
+    # WebSocket connection management
+    # ------------------------------------------------------------------
+
+    async def _connect(self):
+        """Connect to the TTS endpoint and start receiving."""
+        await super()._connect()
+        await self._connect_websocket()
+        if self._websocket and not self._receive_task:
+            self._receive_task = self.create_task(self._receive_task_handler(self._report_error))
+
+    async def _disconnect(self):
+        """Disconnect and clean up background tasks."""
+        await super()._disconnect()
+        if self._receive_task:
+            await self.cancel_task(self._receive_task, timeout=1.0)
+            self._receive_task = None
+        await self._disconnect_websocket()
+
+    async def _connect_websocket(self):
+        """Establish the WebSocket connection to the Together AI TTS endpoint."""
+        try:
+            if self._websocket and self._websocket.state is State.OPEN:
+                return
+
+            ws_url = self._build_websocket_url()
+            logger.debug(f"Connecting to Together AI TTS: {ws_url}")
+
+            headers = {"Authorization": f"Bearer {self._api_key}"}
+            self._websocket = await websocket_connect(ws_url, additional_headers=headers)
+            await self._call_event_handler("on_connected")
+
+            # Ensure voice is set on the server side
+            try:
+                voice_update_msg = {
+                    "type": "tts_session.updated",
+                    "session": {"voice": self._settings.voice},
+                }
+                await self._websocket.send(json.dumps(voice_update_msg))
+                logger.debug(f"Sent initial voice setting to WebSocket: {self._settings.voice}")
+            except Exception as e:
+                logger.error(f"Error sending initial voice setting: {e}")
+
+            logger.debug("Connected to Together AI TTS")
+
+        except Exception as e:
+            await self.push_error(
+                error_msg=f"Error connecting to Together AI TTS: {e}",
+                exception=e,
+            )
+            self._websocket = None
+
+    async def _disconnect_websocket(self):
+        """Close the WebSocket connection."""
+        try:
+            if self._websocket:
+                await self._websocket.close()
+        except Exception as e:
+            await self.push_error(
+                error_msg=f"Error disconnecting: {e}",
+                exception=e,
+            )
+        finally:
+            self._websocket = None
+            self._session_id = None
+            await self._call_event_handler("on_disconnected")
+
+    # ------------------------------------------------------------------
+    # Client events
+    # ------------------------------------------------------------------
+
+    async def _ws_send(self, message: dict):
+        """Send a JSON message over the WebSocket.
+
+        Args:
+            message: The message dict to serialize and send.
+        """
+        if not self._disconnecting:
+            await self.send_with_retry(json.dumps(message), self._report_error)
+
+    async def flush_audio(self, context_id: Optional[str] = None):
+        """Flush any pending audio and finalize the current context.
+
+        If all server-side commits have been completed, closes the audio context
+        immediately. Otherwise, marks the context for deferred closure so
+        ``_handle_audio_done`` can close it when the last commit finishes.
+
+        Args:
+            context_id: Pipecat TTS context to flush.
+        """
+        ctx_id = context_id or self._context_id
+        if not ctx_id or not self.audio_context_available(ctx_id):
+            return
+        logger.trace(f"{self}: flushing audio (context_id={ctx_id})")
+        if self._pending_commits == 0:
+            await self.append_to_audio_context(ctx_id, TTSStoppedFrame(context_id=ctx_id))
+            await self.remove_audio_context(ctx_id)
+        else:
+            self._flush_context_id = ctx_id
+
+    # ------------------------------------------------------------------
+    # Server event handling
+    # ------------------------------------------------------------------
+
+    async def _receive_messages(self):
+        """Receive and dispatch server events from the TTS session.
+
+        Called by ``WebsocketService._receive_task_handler`` which wraps
+        this method with automatic reconnection on connection errors.
+        """
+        async for message in self._websocket:
+            if isinstance(message, bytes):
+                message = message.decode("utf-8")
+            elif not isinstance(message, str):
+                continue
+
+            try:
+                evt = json.loads(message)
+            except json.JSONDecodeError:
+                logger.warning(f"{self} failed to parse WebSocket message")
+                continue
+
+            evt_type = evt.get("type", "")
+
+            if evt_type == "session.created":
+                await self._handle_session_created(evt)
+            elif evt_type == "session.updated":
+                await self._handle_session_updated(evt)
+            elif evt_type == "conversation.item.input_text.received":
+                text = evt.get("text", "")
+                logger.debug(f"{self} text received: {text[:50]}{'...' if len(text) > 50 else ''}")
+            elif evt_type == "conversation.item.audio_output.delta":
+                await self._handle_audio_delta(evt)
+            elif evt_type == "conversation.item.audio_output.done":
+                await self._handle_audio_done(evt)
+            elif evt_type == "conversation.item.tts.failed":
+                await self._handle_tts_failed(evt)
+            elif evt_type == "error":
+                await self._handle_error(evt)
+            else:
+                logger.trace(f"{self} unhandled event: {evt_type}")
+
+    async def _handle_session_created(self, evt: dict):
+        """Handle ``session.created`` event.
+
+        Args:
+            evt: The session created event from the server.
+        """
+        session = evt.get("session", {})
+        self._session_id = session.get("id")
+        logger.debug(f"{self} session created: {self._session_id}")
+
+    async def _handle_session_updated(self, evt: dict):
+        """Handle ``session.updated`` event.
+
+        Args:
+            evt: The session updated event from the server.
+        """
+        session = evt.get("session", {})
+        if "voice" in session:
+            updated_voice = session.get("voice")
+            logger.debug(f"{self} voice updated to: {updated_voice}")
+
+    async def _handle_audio_delta(self, evt: dict):
+        """Handle an audio output delta containing a chunk of synthesized audio.
+
+        Args:
+            evt: The delta event from the server.
+        """
+        if not self._context_id or not self.audio_context_available(self._context_id):
+            return
+        delta = evt.get("delta")
+        if delta:
+            try:
+                audio_chunk = base64.b64decode(delta)
+                frame = TTSAudioRawFrame(
+                    audio=audio_chunk,
+                    sample_rate=self.sample_rate,
+                    num_channels=1,
+                    context_id=self._context_id,
+                )
+                await self.append_to_audio_context(self._context_id, frame)
+            except Exception as e:
+                logger.error(f"{self} error processing audio delta: {e}")
+
+    async def _handle_audio_done(self, evt: dict):
+        """Handle audio output completion for a speech segment.
+
+        Decrements the pending commit counter and closes the audio context
+        if a flush was requested and this was the last pending commit.
+
+        Args:
+            evt: The done event from the server.
+        """
+        if not self._context_id or not self.audio_context_available(self._context_id):
+            return
+        item_id = evt.get("item_id")
+        logger.debug(f"{self} audio generation complete for: {item_id}")
+        self._pending_commits = max(0, self._pending_commits - 1)
+        await self._maybe_close_context()
+
+    async def _handle_tts_failed(self, evt: dict):
+        """Handle a TTS failure.
+
+        Args:
+            evt: The failed event containing error details.
+        """
+        error = evt.get("error", {})
+        self._pending_commits = max(0, self._pending_commits - 1)
+        await self.push_error(error_msg=f"TTS error: {error}")
+        await self._maybe_close_context()
+
+    async def _handle_error(self, evt: dict):
+        """Handle a fatal error from the TTS session.
+
+        Raises an exception so that ``WebsocketService`` can decide
+        whether to attempt reconnection.
+
+        Args:
+            evt: The error event.
+        """
+        error = evt.get("error", {})
+        error_msg = error.get("message", "Unknown error")
+        error_code = error.get("code", "")
+        msg = f"Together AI TTS error [{error_code}]: {error_msg}"
+        await self.push_error(error_msg=msg)
+        raise Exception(msg)
+
+    async def _maybe_close_context(self):
+        """Close the audio context if a flush was requested and no commits remain."""
+        if self._pending_commits == 0 and self._flush_context_id:
+            ctx_id = self._flush_context_id
+            self._flush_context_id = None
+            await self.append_to_audio_context(ctx_id, TTSStoppedFrame(context_id=ctx_id))
+            await self.remove_audio_context(ctx_id)
+
+    # ------------------------------------------------------------------
+    # Interruption handling
+    # ------------------------------------------------------------------
+
+    async def on_audio_context_interrupted(self, context_id: str):
+        """Cancel current generation when the bot is interrupted.
+
+        Args:
+            context_id: The ID of the audio context that was interrupted.
+        """
+        await self.stop_all_metrics()
+        self._pending_commits = 0
+        self._flush_context_id = None
+        await self._ws_send({"type": "input_text_buffer.clear"})
+
+    # ------------------------------------------------------------------
+    # TTS generation
+    # ------------------------------------------------------------------
+
+    @traced_tts
+    async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
+        """Generate speech from text using Together AI's streaming API.
+
+        Audio frames are delivered asynchronously via the WebSocket receive
+        loop and routed through the audio context managed by the base class.
+
+        Args:
+            text: The text to synthesize into speech.
+            context_id: The context ID for tracking audio frames.
+
+        Yields:
+            Frame: None (audio arrives via WebSocket callbacks).
+        """
+        logger.debug(f"{self}: Generating TTS [{text}]")
+
+        try:
+            if not self._websocket or self._websocket.state is not State.OPEN:
+                await self._connect()
+                if not self._websocket or self._websocket.state is not State.OPEN:
+                    logger.error(f"{self} failed to connect to WebSocket")
+                    yield TTSStoppedFrame(context_id=context_id)
+                    return
+
+            self._context_id = context_id
+            self._pending_commits += 1
+
+            try:
+                await self._ws_send({"type": "input_text_buffer.append", "text": text})
+                await self._ws_send({"type": "input_text_buffer.commit"})
+                await self.start_tts_usage_metrics(text)
+            except Exception as e:
+                logger.error(f"{self} error sending message: {e}")
+                self._pending_commits -= 1
+                yield TTSStoppedFrame(context_id=context_id)
+                await self._disconnect()
+                await self._connect()
+                return
+
+            yield None
+
+        except Exception as e:
+            logger.error(f"{self} exception: {e}")
+            await self.push_error(error_msg=f"Error generating TTS: {e}", exception=e)
+            yield TTSStoppedFrame(context_id=context_id)

src/pipecat/services/tts_service.py

@@ -43,6 +43,7 @@ from pipecat.frames.frames import (
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     StartFrame,
+    SystemFrame,
     TextFrame,
     TranscriptionFrame,
     TTSAudioRawFrame,
@@ -248,6 +249,18 @@ class TTSService(AIService):
         # Convert Language enum to service-specific format at init time.
         # Runtime updates are handled by _update_settings(), but init-time
         # settings bypass that path and need explicit conversion.
+        # Raw strings (e.g. "de-DE") are first converted to Language enums
+        # so they go through the same resolution logic.
+        if isinstance(self._settings.language, str) and not isinstance(
+            self._settings.language, Language
+        ):
+            try:
+                self._settings.language = Language(self._settings.language)
+            except ValueError:
+                logger.warning(
+                    f"Language string '{self._settings.language}' is not a recognized "
+                    f"Language code. It will be passed to the service as-is."
+                )
         if isinstance(self._settings.language, Language):
             converted = self.language_to_service_language(self._settings.language)
             if converted is not None:
@@ -545,9 +558,9 @@ class TTSService(AIService):
         """
         await super().stop(frame)
         if self._audio_context_task:
-            # Indicate no more audio contexts are available; this will end the
-            # task cleanly after all contexts have been processed.
-            await self._contexts_queue.put(None)
+            # Sentinel None shuts down the serialization queue once all
+            # pending contexts and frames have been processed.
+            await self._serialization_queue.put(None)
             await self._audio_context_task
             self._audio_context_task = None
         if self._stop_frame_task:
@@ -610,7 +623,20 @@ class TTSService(AIService):
         Returns:
             Dict mapping changed field names to their previous values.
         """
-        # Translate language *before* applying so the stored value is canonical
+        # Translate language *before* applying so the stored value is canonical.
+        # Raw strings are first converted to Language enums for proper resolution.
+        if (
+            is_given(delta.language)
+            and isinstance(delta.language, str)
+            and not isinstance(delta.language, Language)
+        ):
+            try:
+                delta.language = Language(delta.language)
+            except ValueError:
+                logger.warning(
+                    f"Language string '{delta.language}' is not a recognized "
+                    f"Language code. It will be passed to the service as-is."
+                )
         if is_given(delta.language) and isinstance(delta.language, Language):
             converted = self.language_to_service_language(delta.language)
             if converted is not None:
@@ -695,11 +721,6 @@ class TTSService(AIService):
             self._turn_context_id = self.create_context_id()
             await self.push_frame(frame, direction)
         elif isinstance(frame, (LLMFullResponseEndFrame, EndFrame)):
-            # We pause processing incoming frames if the LLM response included
-            # text (it might be that it's only a function calling response). We
-            # pause to avoid audio overlapping.
-            await self._maybe_pause_frame_processing()
-
             # Flush any remaining text (including text waiting for lookahead)
             remaining = await self._text_aggregator.flush()
             # Stop the aggregation metric (no-op if already stopped on first sentence).
@@ -707,6 +728,11 @@ class TTSService(AIService):
             if remaining:
                 await self._push_tts_frames(AggregatedTextFrame(remaining.text, remaining.type))
 
+            # We pause processing incoming frames if the LLM response included
+            # text (it might be that it's only a function calling response). We
+            # pause to avoid audio overlapping.
+            await self._maybe_pause_frame_processing()
+
             # Log accumulated streamed text and emit aggregated usage metric.
             if self._streamed_text:
                 logger.debug(f"{self}: Generating TTS [{self._streamed_text}]")
@@ -766,7 +792,15 @@ class TTSService(AIService):
             await self._maybe_resume_frame_processing()
             await self.push_frame(frame, direction)
         else:
-            await self.push_frame(frame, direction)
+            if direction == FrameDirection.DOWNSTREAM and not isinstance(frame, SystemFrame):
+                # Route non-system downstream frames through the serialization queue so they
+                # are emitted in the same order they arrive relative to any audio contexts that
+                # are already queued (e.g. a FooFrame sent right after a TTSSpeakFrame must
+                # not overtake the TTSStartedFrame / TTSAudioRawFrame / TTSStoppedFrame
+                # sequence from that speak frame).
+                await self._serialization_queue.put(frame)
+            else:
+                await self.push_frame(frame, direction)
 
     async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
         """Push a frame downstream with TTS-specific handling.
@@ -969,7 +1003,15 @@ class TTSService(AIService):
         # is spoken, so we set append_to_context to False.
         src_frame.append_to_context = False
         src_frame.context_id = context_id
-        await self.push_frame(src_frame)
+        # Route AggregatedTextFrame through the serialization queue so it is emitted
+        # immediately before the TTSStartedFrame of the audio context it describes,
+        # rather than racing ahead of audio frames from a previous context.
+        if not self.audio_context_available(context_id):
+            await self._serialization_queue.put(src_frame)
+        # Otherwise, if the context already exists, we append the AggregatedTextFrame
+        # to the existing context queue.
+        else:
+            await self.append_to_audio_context(context_id, src_frame)
 
         # Note: Text transformations are meant to only affect the text sent to the TTS for
         # TTS-specific purposes. This allows for explicit TTS modifications (e.g., inserting
@@ -1018,11 +1060,8 @@ class TTSService(AIService):
             # Only override append_to_context if explicitly set
             if append_tts_text_to_context is not None:
                 frame.append_to_context = append_tts_text_to_context
-            # For services using the audio context we are appending to the context, so it preserves the ordering.
-            if self.audio_context_available(context_id):
-                await self.append_to_audio_context(context_id, frame)
-            else:
-                await self.push_frame(frame)
+            # Appending to the context, so it preserves the ordering.
+            await self.append_to_audio_context(context_id, frame)
 
     async def tts_process_generator(
         self, context_id: str, generator: AsyncGenerator[Frame | None, None]
@@ -1178,7 +1217,7 @@ class TTSService(AIService):
         Args:
             context_id: Unique identifier for the audio context.
         """
-        await self._contexts_queue.put(context_id)
+        await self._serialization_queue.put(context_id)
         self._audio_contexts[context_id] = asyncio.Queue()
         logger.trace(f"{self} created audio context {context_id}")
 
@@ -1270,7 +1309,14 @@ class TTSService(AIService):
 
     def _create_audio_context_task(self):
         if not self._audio_context_task:
-            self._contexts_queue: asyncio.Queue = asyncio.Queue()
+            # Single FIFO queue that serializes everything the TTS service emits downstream.
+            # Items can be:
+            #   str   – an audio context ID: process the per-context audio queue in full before
+            #           moving on (see _handle_audio_context).
+            #   Frame – a non-system downstream frame (e.g. AggregatedTextFrame, FooFrame) that
+            #           must be emitted in-order relative to surrounding audio contexts.
+            #   None  – shutdown sentinel (sent by stop()).
+            self._serialization_queue: asyncio.Queue = asyncio.Queue()
             self._audio_contexts: Dict[str, asyncio.Queue] = {}
             self._audio_context_task = self.create_task(self._audio_context_task_handler())
 
@@ -1280,13 +1326,26 @@ class TTSService(AIService):
             self._audio_context_task = None
 
     async def _audio_context_task_handler(self):
-        """In this task we process audio contexts in order."""
+        """Drain the serialization queue, preserving downstream frame order.
+
+        The queue carries three kinds of items (see _create_audio_context_task):
+
+        * str  – audio context ID: block until all audio for that context has been
+                 pushed downstream, then call on_audio_context_completed().
+        * Frame – a non-system downstream frame that must be emitted at this exact
+                  position in the output stream (e.g. AggregatedTextFrame preceding
+                  its audio, or an arbitrary frame that arrived between two speak frames).
+        * None – shutdown sentinel; exit the loop once reached.
+        """
         running = True
         while running:
-            context_id = await self._contexts_queue.get()
-            self._playing_context_id = context_id
+            context_value = await self._serialization_queue.get()
+            if isinstance(context_value, Frame):
+                await self.push_frame(context_value)
+            elif isinstance(context_value, str):
+                context_id = context_value
+                self._playing_context_id = context_id
 
-            if context_id:
                 # Process the audio context until the context doesn't have more
                 # audio available (i.e. we find None).
                 await self._handle_audio_context(context_id)
@@ -1298,7 +1357,7 @@ class TTSService(AIService):
             else:
                 running = False
 
-            self._contexts_queue.task_done()
+            self._serialization_queue.task_done()
 
     async def _handle_audio_context(self, context_id: str):
         """Process items from an audio context queue until it is exhausted."""

src/pipecat/transcriptions/language.py

@@ -631,13 +631,13 @@ def resolve_language(
         return result
 
     # Not in map - fall back with warning
-    lang_str = str(language.value)
+    lang_str = str(language)
 
     if use_base_code:
         # Extract base code (e.g., "en" from "en-US")
         base_code = lang_str.split("-")[0].lower()
-        logger.warning(f"Language {language.value} not verified. Using base code '{base_code}'.")
+        logger.warning(f"Language {language} not verified. Using base code '{base_code}'.")
         return base_code
     else:
-        logger.warning(f"Language {language.value} not verified. Using '{lang_str}'.")
+        logger.warning(f"Language {language} not verified. Using '{lang_str}'.")
         return lang_str

src/pipecat/transports/base_output.py

@@ -569,7 +569,11 @@ class BaseOutputTransport(FrameProcessor):
             if not self._params.video_out_enabled:
                 return
 
-            if self._params.video_out_is_live and isinstance(frame, OutputImageRawFrame):
+            if isinstance(frame, OutputImageRawFrame) and frame.sync_with_audio:
+                # Route through the audio queue so the image is only
+                # displayed after all preceding audio has been sent.
+                await self._audio_queue.put(frame)
+            elif self._params.video_out_is_live and isinstance(frame, OutputImageRawFrame):
                 await self._video_queue.put(frame)
             elif isinstance(frame, OutputImageRawFrame):
                 await self._set_video_image(frame)

src/pipecat/transports/daily/utils.py

@@ -89,7 +89,7 @@ class DailyRoomProperties(BaseModel):
         enable_emoji_reactions: Whether emoji reactions are enabled.
         eject_at_room_exp: Whether to remove participants when room expires.
         enable_dialout: Whether SIP dial-out is enabled.
-        enable_recording: Recording settings ('cloud', 'local', 'raw-tracks').
+        enable_recording: Recording settings ('cloud', 'cloud-audio-only', 'local', 'raw-tracks').
         enable_transcription_storage: Whether transcription storage is enabled.
         geo: Geographic region for room.
         max_participants: Maximum number of participants allowed in the room.
@@ -185,7 +185,7 @@ class DailyMeetingTokenProperties(BaseModel):
         enable_screenshare: If True, the user will be able to share their screen.
         start_video_off: If True, the user's video will be turned off when they join the room.
         start_audio_off: If True, the user's audio will be turned off when they join the room.
-        enable_recording: Recording settings for the token. Must be one of 'cloud', 'local' or 'raw-tracks'.
+        enable_recording: Recording settings for the token. Must be one of 'cloud', 'cloud-audio-only', 'local' or 'raw-tracks'.
         enable_prejoin_ui: If True, the user will see the prejoin UI before joining the room.
         start_cloud_recording: Start cloud recording when the user joins the room.
         permissions: Specifies the initial default permissions for a non-meeting-owner participant.

src/pipecat/transports/livekit/transport.py

@@ -388,7 +388,16 @@ class LiveKitTransportClient:
             await self._audio_source.capture_frame(audio_frame)
             return True
         except Exception as e:
-            logger.error(f"Error publishing audio: {e}")
+            # When using an audio mixer, the base output transport's
+            # with_mixer() generator continuously yields frames (mixed with
+            # background audio) even when no TTS audio is queued. During
+            # interruptions, the audio task is cancelled and recreated, but
+            # there is a brief window where the native LiveKit AudioSource
+            # rejects capture_frame() with an InvalidState error. This is a
+            # transient condition — the mixer will produce a new frame within
+            # milliseconds, so we silently drop these frames.
+            if "InvalidState" not in str(e):
+                logger.error(f"Error publishing audio: {e}")
             return False
 
     def get_participants(self) -> List[str]:

src/pipecat/transports/tavus/transport.py

@@ -241,6 +241,7 @@ class TavusTransportClient:
                 on_dialout_stopped=partial(self._on_handle_callback, "on_dialout_stopped"),
                 on_dialout_error=partial(self._on_handle_callback, "on_dialout_error"),
                 on_dialout_warning=partial(self._on_handle_callback, "on_dialout_warning"),
+                on_dtmf_event=partial(self._on_handle_callback, "on_dtmf_event"),
                 on_participant_joined=self._callbacks.on_participant_joined,
                 on_participant_left=self._callbacks.on_participant_left,
                 on_participant_updated=partial(self._on_handle_callback, "on_participant_updated"),

src/pipecat/turns/types.py

@@ -0,0 +1,24 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Shared result type for user turn strategy frame processing."""
+
+from enum import Enum
+
+
+class ProcessFrameResult(Enum):
+    """Result of processing a frame in a user turn strategy.
+
+    Controls whether the strategy loop in the controller continues to the
+    next strategy or stops early.
+
+    Attributes:
+        CONTINUE: Continue to the next strategy in the loop.
+        STOP: Stop evaluating further strategies for this frame.
+    """
+
+    CONTINUE = "continue"
+    STOP = "stop"

src/pipecat/turns/user_start/__init__.py

@@ -9,6 +9,7 @@ from .external_user_turn_start_strategy import ExternalUserTurnStartStrategy
 from .min_words_user_turn_start_strategy import MinWordsUserTurnStartStrategy
 from .transcription_user_turn_start_strategy import TranscriptionUserTurnStartStrategy
 from .vad_user_turn_start_strategy import VADUserTurnStartStrategy
+from .wake_phrase_user_turn_start_strategy import WakePhraseUserTurnStartStrategy
 
 __all__ = [
     "BaseUserTurnStartStrategy",
@@ -17,4 +18,5 @@ __all__ = [
     "TranscriptionUserTurnStartStrategy",
     "UserTurnStartedParams",
     "VADUserTurnStartStrategy",
+    "WakePhraseUserTurnStartStrategy",
 ]

src/pipecat/turns/user_start/base_user_turn_start_strategy.py

@@ -11,6 +11,7 @@ from typing import Optional, Type
 
 from pipecat.frames.frames import Frame
 from pipecat.processors.frame_processor import FrameDirection
+from pipecat.turns.types import ProcessFrameResult
 from pipecat.utils.asyncio.task_manager import BaseTaskManager
 from pipecat.utils.base_object import BaseObject
 
@@ -76,6 +77,7 @@ class BaseUserTurnStartStrategy(BaseObject):
         self._register_event_handler("on_push_frame", sync=True)
         self._register_event_handler("on_broadcast_frame", sync=True)
         self._register_event_handler("on_user_turn_started", sync=True)
+        self._register_event_handler("on_reset_aggregation", sync=True)
 
     @property
     def task_manager(self) -> BaseTaskManager:
@@ -100,7 +102,7 @@ class BaseUserTurnStartStrategy(BaseObject):
         """Reset the strategy to its initial state."""
         pass
 
-    async def process_frame(self, frame: Frame):
+    async def process_frame(self, frame: Frame) -> ProcessFrameResult:
         """Process an incoming frame.
 
         Subclasses should override this to implement logic that decides whether
@@ -108,6 +110,10 @@ class BaseUserTurnStartStrategy(BaseObject):
 
         Args:
             frame: The frame to be processed.
+
+        Returns:
+            A ProcessFrameResult indicating the outcome. Subclasses that return
+            None are treated as CONTINUE for backward compatibility.
         """
         pass
 
@@ -138,3 +144,7 @@ class BaseUserTurnStartStrategy(BaseObject):
                 enable_user_speaking_frames=self._enable_user_speaking_frames,
             ),
         )
+
+    async def trigger_reset_aggregation(self):
+        """Trigger the `on_reset_aggregation` event."""
+        await self._call_event_handler("on_reset_aggregation")

src/pipecat/turns/user_start/external_user_turn_start_strategy.py

@@ -7,6 +7,7 @@
 """User turn start strategy triggered by externally emitted frames."""
 
 from pipecat.frames.frames import Frame, UserStartedSpeakingFrame
+from pipecat.turns.types import ProcessFrameResult
 from pipecat.turns.user_start.base_user_turn_start_strategy import BaseUserTurnStartStrategy
 
 
@@ -27,13 +28,17 @@ class ExternalUserTurnStartStrategy(BaseUserTurnStartStrategy):
         """
         super().__init__(enable_interruptions=False, enable_user_speaking_frames=False, **kwargs)
 
-    async def process_frame(self, frame: Frame):
+    async def process_frame(self, frame: Frame) -> ProcessFrameResult:
         """Process an incoming frame to detect user turn start.
 
         Args:
             frame: The frame to be analyzed.
-        """
-        await super().process_frame(frame)
 
+        Returns:
+            STOP if a user started speaking frame was received, CONTINUE otherwise.
+        """
         if isinstance(frame, UserStartedSpeakingFrame):
             await self.trigger_user_turn_started()
+            return ProcessFrameResult.STOP
+
+        return ProcessFrameResult.CONTINUE

src/pipecat/turns/user_start/min_words_user_turn_start_strategy.py

@@ -15,6 +15,7 @@ from pipecat.frames.frames import (
     InterimTranscriptionFrame,
     TranscriptionFrame,
 )
+from pipecat.turns.types import ProcessFrameResult
 from pipecat.turns.user_start.base_user_turn_start_strategy import BaseUserTurnStartStrategy
 
 
@@ -47,7 +48,7 @@ class MinWordsUserTurnStartStrategy(BaseUserTurnStartStrategy):
         await super().reset()
         self._bot_speaking = False
 
-    async def process_frame(self, frame: Frame):
+    async def process_frame(self, frame: Frame) -> ProcessFrameResult:
         """Process an incoming frame to detect the start of a user turn.
 
         This method updates internal state based on transcription frames and
@@ -55,17 +56,20 @@ class MinWordsUserTurnStartStrategy(BaseUserTurnStartStrategy):
 
         Args:
             frame: The frame to be analyzed.
-        """
-        await super().process_frame(frame)
 
+        Returns:
+            STOP if the minimum word count was reached, CONTINUE otherwise.
+        """
         if isinstance(frame, BotStartedSpeakingFrame):
             await self._handle_bot_started_speaking(frame)
         elif isinstance(frame, BotStoppedSpeakingFrame):
             await self._handle_bot_stopped_speaking(frame)
         elif isinstance(frame, TranscriptionFrame):
-            await self._handle_transcription(frame)
+            return await self._handle_transcription(frame)
         elif isinstance(frame, InterimTranscriptionFrame) and self._use_interim:
-            await self._handle_transcription(frame)
+            return await self._handle_transcription(frame)
+
+        return ProcessFrameResult.CONTINUE
 
     async def _handle_bot_started_speaking(self, frame: BotStartedSpeakingFrame):
         """Handle bot started speaking frame.
@@ -87,11 +91,16 @@ class MinWordsUserTurnStartStrategy(BaseUserTurnStartStrategy):
         """
         self._bot_speaking = False
 
-    async def _handle_transcription(self, frame: TranscriptionFrame | InterimTranscriptionFrame):
-        """Handle a completed transcription frame and check word count.
+    async def _handle_transcription(
+        self, frame: TranscriptionFrame | InterimTranscriptionFrame
+    ) -> ProcessFrameResult:
+        """Handle a transcription frame and check word count.
 
         Args:
             frame: The transcription frame to be processed.
+
+        Returns:
+            STOP if the minimum word count was reached, CONTINUE otherwise.
         """
         min_words = self._min_words if self._bot_speaking else 1
 
@@ -106,3 +115,7 @@ class MinWordsUserTurnStartStrategy(BaseUserTurnStartStrategy):
 
         if should_trigger:
             await self.trigger_user_turn_started()
+            return ProcessFrameResult.STOP
+        await self.trigger_reset_aggregation()
+
+        return ProcessFrameResult.CONTINUE

src/pipecat/turns/user_start/transcription_user_turn_start_strategy.py

@@ -7,6 +7,7 @@
 """User turn start strategy based on transcriptions."""
 
 from pipecat.frames.frames import Frame, InterimTranscriptionFrame, TranscriptionFrame
+from pipecat.turns.types import ProcessFrameResult
 from pipecat.turns.user_start.base_user_turn_start_strategy import BaseUserTurnStartStrategy
 
 
@@ -25,15 +26,20 @@ class TranscriptionUserTurnStartStrategy(BaseUserTurnStartStrategy):
         super().__init__(**kwargs)
         self._use_interim = use_interim
 
-    async def process_frame(self, frame: Frame):
+    async def process_frame(self, frame: Frame) -> ProcessFrameResult:
         """Process an incoming frame to detect the start of a user turn.
 
         Args:
             frame: The frame to be processed.
-        """
-        await super().process_frame(frame)
 
+        Returns:
+            STOP if a transcription was received, CONTINUE otherwise.
+        """
         if isinstance(frame, InterimTranscriptionFrame) and self._use_interim:
             await self.trigger_user_turn_started()
+            return ProcessFrameResult.STOP
         elif isinstance(frame, TranscriptionFrame):
             await self.trigger_user_turn_started()
+            return ProcessFrameResult.STOP
+
+        return ProcessFrameResult.CONTINUE

src/pipecat/turns/user_start/vad_user_turn_start_strategy.py

@@ -7,6 +7,7 @@
 """User turn start strategy based on VAD events."""
 
 from pipecat.frames.frames import Frame, VADUserStartedSpeakingFrame
+from pipecat.turns.types import ProcessFrameResult
 from pipecat.turns.user_start.base_user_turn_start_strategy import BaseUserTurnStartStrategy
 
 
@@ -18,13 +19,17 @@ class VADUserTurnStartStrategy(BaseUserTurnStartStrategy):
 
     """
 
-    async def process_frame(self, frame: Frame):
+    async def process_frame(self, frame: Frame) -> ProcessFrameResult:
         """Process an incoming frame to detect user turn start.
 
         Args:
             frame: The frame to be analyzed.
-        """
-        await super().process_frame(frame)
 
+        Returns:
+            STOP if the user started speaking, CONTINUE otherwise.
+        """
         if isinstance(frame, VADUserStartedSpeakingFrame):
             await self.trigger_user_turn_started()
+            return ProcessFrameResult.STOP
+
+        return ProcessFrameResult.CONTINUE

src/pipecat/turns/user_start/wake_phrase_user_turn_start_strategy.py

@@ -0,0 +1,281 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""User turn start strategy that gates interaction behind wake phrase detection."""
+
+import asyncio
+import enum
+import re
+from typing import List, Optional
+
+from loguru import logger
+
+from pipecat.frames.frames import (
+    BotSpeakingFrame,
+    Frame,
+    TranscriptionFrame,
+    UserSpeakingFrame,
+    VADUserStartedSpeakingFrame,
+)
+from pipecat.turns.types import ProcessFrameResult
+from pipecat.turns.user_start.base_user_turn_start_strategy import BaseUserTurnStartStrategy
+from pipecat.utils.asyncio.task_manager import BaseTaskManager
+
+
+class _WakeState(enum.Enum):
+    """Internal state for wake phrase detection."""
+
+    IDLE = "idle"
+    AWAKE = "awake"
+
+
+class WakePhraseUserTurnStartStrategy(BaseUserTurnStartStrategy):
+    """User turn start strategy that requires a wake phrase before interaction.
+
+    Blocks subsequent strategies until a wake phrase is detected in a final
+    transcription. After detection, allows interaction for a configurable
+    timeout period before requiring the wake phrase again. Use
+    ``single_activation=True`` to require the wake phrase before every turn.
+
+    This strategy should be placed first in the start strategies list.
+
+    Event handlers available:
+
+    - on_wake_phrase_detected: Called when a wake phrase is matched.
+    - on_wake_phrase_timeout: Called when the inactivity timeout expires
+      (timeout mode only).
+
+    Example::
+
+        # Timeout mode (default): wake phrase unlocks interaction for 10s
+        strategy = WakePhraseUserTurnStartStrategy(
+            phrases=["hey pipecat", "ok pipecat"],
+            timeout=10.0,
+        )
+
+        # Single activation: wake phrase required before every turn
+        strategy = WakePhraseUserTurnStartStrategy(
+            phrases=["hey pipecat"],
+            single_activation=True,
+        )
+
+        @strategy.event_handler("on_wake_phrase_detected")
+        async def on_wake_phrase_detected(strategy, phrase):
+            ...
+
+        @strategy.event_handler("on_wake_phrase_timeout")
+        async def on_wake_phrase_timeout(strategy):
+            ...
+
+    Args:
+        phrases: List of wake phrases to detect.
+        timeout: Inactivity timeout in seconds before returning to IDLE.
+            In timeout mode, the timer resets on activity (user, bot speech).
+            In single activation mode, acts as a keepalive window — the strategy
+            stays AWAKE for this duration after wake phrase detection, allowing
+            the current turn to complete before returning to IDLE.
+        single_activation: If True, the wake phrase is required before every
+            turn. The strategy returns to IDLE after each turn completes.
+        **kwargs: Additional keyword arguments passed to parent.
+    """
+
+    def __init__(
+        self,
+        *,
+        phrases: List[str],
+        timeout: float = 10.0,
+        single_activation: bool = False,
+        **kwargs,
+    ):
+        """Initialize the wake phrase user turn start strategy.
+
+        Args:
+            phrases: List of wake phrases to detect.
+            timeout: Inactivity timeout in seconds before returning to IDLE.
+                In timeout mode, the timer resets on activity. In single activation
+                mode, acts as a keepalive window after wake phrase detection.
+            single_activation: If True, the wake phrase is required before every
+                turn. The strategy returns to IDLE after each turn completes.
+            **kwargs: Additional keyword arguments passed to parent.
+        """
+        super().__init__(**kwargs)
+        self._phrases = phrases
+        self._timeout = timeout
+        self._single_activation = single_activation
+
+        self._patterns: List[re.Pattern] = []
+        for phrase in phrases:
+            pattern = re.compile(
+                r"\b" + r"\s*".join(re.escape(word) for word in phrase.split()) + r"\b",
+                re.IGNORECASE,
+            )
+            self._patterns.append(pattern)
+
+        self._state = _WakeState.IDLE
+        self._accumulated_text = ""
+
+        self._timeout_event = asyncio.Event()
+        self._timeout_task: Optional[asyncio.Task] = None
+
+        self._register_event_handler("on_wake_phrase_detected")
+        self._register_event_handler("on_wake_phrase_timeout")
+
+    @property
+    def state(self) -> _WakeState:
+        """Returns the current wake state."""
+        return self._state
+
+    async def setup(self, task_manager: BaseTaskManager):
+        """Initialize the strategy with the given task manager.
+
+        Args:
+            task_manager: The task manager to be associated with this instance.
+        """
+        await super().setup(task_manager)
+        if not self._timeout_task:
+            self._timeout_task = self.task_manager.create_task(
+                self._timeout_task_handler(),
+                f"{self}::_timeout_task_handler",
+            )
+
+    async def cleanup(self):
+        """Cleanup the strategy."""
+        await super().cleanup()
+        if self._timeout_task:
+            await self.task_manager.cancel_task(self._timeout_task)
+            self._timeout_task = None
+
+    async def reset(self):
+        """Reset the strategy.
+
+        In timeout mode, preserves state and refreshes timeout since reset
+        means a turn started (activity). In single activation mode, does
+        nothing — the keepalive timeout (started when the wake phrase was
+        detected) handles the transition back to IDLE.
+        """
+        await super().reset()
+        if self._state == _WakeState.AWAKE:
+            if not self._single_activation:
+                self._refresh_timeout()
+
+    async def process_frame(self, frame: Frame) -> ProcessFrameResult:
+        """Process an incoming frame for wake phrase detection or passthrough.
+
+        Args:
+            frame: The frame to be processed.
+
+        Returns:
+            STOP when the wake phrase is detected or when in IDLE state
+            (blocks subsequent strategies), CONTINUE when in AWAKE state
+            (allows subsequent strategies to proceed).
+        """
+        await super().process_frame(frame)
+
+        if self._state == _WakeState.IDLE:
+            return await self._process_idle(frame)
+        else:
+            return await self._process_awake(frame)
+
+    async def _process_idle(self, frame: Frame) -> ProcessFrameResult:
+        """Process a frame while in IDLE state.
+
+        Only final ``TranscriptionFrame`` instances are checked for wake phrase
+        matches. When a match is found, a user turn start is triggered.
+        Transcription frames that don't match have their text cleared so that
+        pre-wake-phrase speech is not added to the LLM context. All frames
+        return STOP to block subsequent strategies.
+        """
+        if isinstance(frame, TranscriptionFrame):
+            if self._check_wake_phrase(frame.text):
+                await self.trigger_user_turn_started()
+                return ProcessFrameResult.STOP
+            await self.trigger_reset_aggregation()
+
+        return ProcessFrameResult.STOP
+
+    async def _process_awake(self, frame: Frame) -> ProcessFrameResult:
+        """Process a frame while in AWAKE state.
+
+        Refreshes the timeout on activity frames (timeout mode only). Returns
+        CONTINUE so subsequent strategies can process the frame.
+        """
+        if not self._single_activation:
+            if isinstance(frame, (UserSpeakingFrame, BotSpeakingFrame)):
+                self._refresh_timeout()
+            elif isinstance(frame, TranscriptionFrame):
+                self._refresh_timeout()
+            elif isinstance(frame, VADUserStartedSpeakingFrame):
+                self._refresh_timeout()
+
+        return ProcessFrameResult.CONTINUE
+
+    @staticmethod
+    def _strip_punctuation(text: str) -> str:
+        """Strip punctuation from text, keeping only letters, digits, and whitespace."""
+        return re.sub(r"[^\w\s]", "", text)
+
+    def _check_wake_phrase(self, text: str) -> bool:
+        """Check if the accumulated text contains a wake phrase.
+
+        Punctuation is stripped before matching so that STT output like
+        "Hey, Pipecat!" still matches the phrase "hey pipecat".
+
+        Args:
+            text: New transcription text to append and check.
+
+        Returns:
+            True if a wake phrase was found, False otherwise.
+        """
+        self._accumulated_text += " " + self._strip_punctuation(text)
+        # Cap accumulated text to prevent unbounded growth.
+        if len(self._accumulated_text) > 250:
+            self._accumulated_text = self._accumulated_text[-250:]
+
+        for i, pattern in enumerate(self._patterns):
+            if pattern.search(self._accumulated_text):
+                phrase = self._phrases[i]
+                logger.debug(f"{self} wake phrase detected: {phrase!r}")
+                self._transition_to_awake(phrase)
+                return True
+
+        return False
+
+    def _transition_to_awake(self, phrase: str):
+        """Transition from IDLE to AWAKE state."""
+        self._state = _WakeState.AWAKE
+        self._accumulated_text = ""
+        self._refresh_timeout()
+        self.task_manager.create_task(
+            self._call_event_handler("on_wake_phrase_detected", phrase),
+            f"{self}::on_wake_phrase_detected",
+        )
+
+    def _transition_to_idle(self):
+        """Transition from AWAKE to IDLE state."""
+        logger.debug(f"{self} wake phrase timeout, returning to IDLE")
+        self._state = _WakeState.IDLE
+        self._accumulated_text = ""
+        self.task_manager.create_task(
+            self._call_event_handler("on_wake_phrase_timeout"),
+            f"{self}::on_wake_phrase_timeout",
+        )
+
+    def _refresh_timeout(self):
+        """Refresh the inactivity timeout."""
+        self._timeout_event.set()
+
+    async def _timeout_task_handler(self):
+        """Background task that monitors inactivity timeout."""
+        while True:
+            try:
+                await asyncio.wait_for(
+                    self._timeout_event.wait(),
+                    timeout=self._timeout,
+                )
+                self._timeout_event.clear()
+            except asyncio.TimeoutError:
+                if self._state == _WakeState.AWAKE:
+                    self._transition_to_idle()

src/pipecat/turns/user_stop/base_user_turn_stop_strategy.py

@@ -11,6 +11,7 @@ from typing import Optional, Type
 
 from pipecat.frames.frames import Frame
 from pipecat.processors.frame_processor import FrameDirection
+from pipecat.turns.types import ProcessFrameResult
 from pipecat.utils.asyncio.task_manager import BaseTaskManager
 from pipecat.utils.base_object import BaseObject
 
@@ -89,7 +90,7 @@ class BaseUserTurnStopStrategy(BaseObject):
         """Reset the strategy to its initial state."""
         pass
 
-    async def process_frame(self, frame: Frame):
+    async def process_frame(self, frame: Frame) -> ProcessFrameResult:
         """Process an incoming frame to decide whether the user stopped speaking.
 
         Subclasses should override this to implement logic that decides whether
@@ -97,6 +98,10 @@ class BaseUserTurnStopStrategy(BaseObject):
 
         Args:
             frame: The frame to be analyzed.
+
+        Returns:
+            A ProcessFrameResult indicating the outcome. Subclasses that return
+            None are treated as CONTINUE for backward compatibility.
         """
         pass
 

src/pipecat/turns/user_stop/external_user_turn_stop_strategy.py

@@ -16,6 +16,7 @@ from pipecat.frames.frames import (
     UserStartedSpeakingFrame,
     UserStoppedSpeakingFrame,
 )
+from pipecat.turns.types import ProcessFrameResult
 from pipecat.turns.user_stop.base_user_turn_stop_strategy import BaseUserTurnStopStrategy
 from pipecat.utils.asyncio.task_manager import BaseTaskManager
 
@@ -69,7 +70,7 @@ class ExternalUserTurnStopStrategy(BaseUserTurnStopStrategy):
             await self.task_manager.cancel_task(self._task)
             self._task = None
 
-    async def process_frame(self, frame: Frame):
+    async def process_frame(self, frame: Frame) -> ProcessFrameResult:
         """Process an incoming frame to update strategy state.
 
         Updates internal transcription text and VAD state. The user end turn
@@ -78,6 +79,8 @@ class ExternalUserTurnStopStrategy(BaseUserTurnStopStrategy):
         Args:
             frame: The frame to be analyzed.
 
+        Returns:
+            Always returns CONTINUE so subsequent stop strategies are evaluated.
         """
         if isinstance(frame, UserStartedSpeakingFrame):
             await self._handle_user_started_speaking(frame)
@@ -88,6 +91,8 @@ class ExternalUserTurnStopStrategy(BaseUserTurnStopStrategy):
         elif isinstance(frame, TranscriptionFrame):
             await self._handle_transcription(frame)
 
+        return ProcessFrameResult.CONTINUE
+
     async def _handle_user_started_speaking(self, _: UserStartedSpeakingFrame):
         """Handle when the external service indicates the user is speaking."""
         self._user_speaking = True

src/pipecat/turns/user_stop/speech_timeout_user_turn_stop_strategy.py

@@ -17,6 +17,7 @@ from pipecat.frames.frames import (
     VADUserStartedSpeakingFrame,
     VADUserStoppedSpeakingFrame,
 )
+from pipecat.turns.types import ProcessFrameResult
 from pipecat.turns.user_stop.base_user_turn_stop_strategy import BaseUserTurnStopStrategy
 from pipecat.utils.asyncio.task_manager import BaseTaskManager
 
@@ -64,6 +65,9 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy):
         self._vad_user_speaking = False
         self._transcript_finalized = False
         self._vad_stopped_time = None
+        if self._timeout_task:
+            await self.task_manager.cancel_task(self._timeout_task)
+            self._timeout_task = None
 
     async def setup(self, task_manager: BaseTaskManager):
         """Initialize the strategy with the given task manager.
@@ -80,7 +84,7 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy):
             await self.task_manager.cancel_task(self._timeout_task)
             self._timeout_task = None
 
-    async def process_frame(self, frame: Frame):
+    async def process_frame(self, frame: Frame) -> ProcessFrameResult:
         """Process an incoming frame to update strategy state.
 
         Updates internal transcription text and VAD state. The user end turn
@@ -89,6 +93,8 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy):
         Args:
             frame: The frame to be analyzed.
 
+        Returns:
+            Always returns CONTINUE so subsequent stop strategies are evaluated.
         """
         if isinstance(frame, STTMetadataFrame):
             self._stt_timeout = frame.ttfs_p99_latency
@@ -99,6 +105,8 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy):
         elif isinstance(frame, TranscriptionFrame):
             await self._handle_transcription(frame)
 
+        return ProcessFrameResult.CONTINUE
+
     async def _handle_vad_user_started_speaking(self, _: VADUserStartedSpeakingFrame):
         """Handle when the VAD indicates the user is speaking."""
         self._vad_user_speaking = True

src/pipecat/turns/user_stop/turn_analyzer_user_turn_stop_strategy.py

@@ -22,6 +22,7 @@ from pipecat.frames.frames import (
     VADUserStoppedSpeakingFrame,
 )
 from pipecat.metrics.metrics import MetricsData
+from pipecat.turns.types import ProcessFrameResult
 from pipecat.turns.user_stop.base_user_turn_stop_strategy import BaseUserTurnStopStrategy
 from pipecat.utils.asyncio.task_manager import BaseTaskManager
 
@@ -68,6 +69,9 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy):
         self._vad_user_speaking = False
         self._vad_stopped_time = None
         self._transcript_finalized = False
+        if self._timeout_task:
+            await self.task_manager.cancel_task(self._timeout_task)
+            self._timeout_task = None
 
     async def setup(self, task_manager: BaseTaskManager):
         """Initialize the strategy with the given task manager.
@@ -85,11 +89,14 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy):
             await self.task_manager.cancel_task(self._timeout_task)
             self._timeout_task = None
 
-    async def process_frame(self, frame: Frame):
+    async def process_frame(self, frame: Frame) -> ProcessFrameResult:
         """Process an incoming frame to update the turn analyzer and strategy state.
 
         Args:
             frame: The frame to be analyzed.
+
+        Returns:
+            Always returns CONTINUE so subsequent stop strategies are evaluated.
         """
         await super().process_frame(frame)
 
@@ -106,6 +113,8 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy):
         elif isinstance(frame, TranscriptionFrame):
             await self._handle_transcription(frame)
 
+        return ProcessFrameResult.CONTINUE
+
     async def _start(self, frame: StartFrame):
         """Process the start frame to configure the turn analyzer."""
         self._turn_analyzer.set_sample_rate(frame.audio_in_sample_rate)

src/pipecat/turns/user_turn_controller.py

@@ -19,7 +19,11 @@ from pipecat.frames.frames import (
     VADUserStoppedSpeakingFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
-from pipecat.turns.user_start import BaseUserTurnStartStrategy, UserTurnStartedParams
+from pipecat.turns.types import ProcessFrameResult
+from pipecat.turns.user_start import (
+    BaseUserTurnStartStrategy,
+    UserTurnStartedParams,
+)
 from pipecat.turns.user_stop import BaseUserTurnStopStrategy, UserTurnStoppedParams
 from pipecat.turns.user_turn_strategies import UserTurnStrategies
 from pipecat.utils.asyncio.task_manager import BaseTaskManager
@@ -94,6 +98,7 @@ class UserTurnController(BaseObject):
         self._register_event_handler("on_user_turn_started", sync=True)
         self._register_event_handler("on_user_turn_stopped", sync=True)
         self._register_event_handler("on_user_turn_stop_timeout", sync=True)
+        self._register_event_handler("on_reset_aggregation", sync=True)
 
     @property
     def task_manager(self) -> BaseTaskManager:
@@ -161,10 +166,14 @@ class UserTurnController(BaseObject):
             await self._handle_transcription(frame)
 
         for strategy in self._user_turn_strategies.start or []:
-            await strategy.process_frame(frame)
+            result = await strategy.process_frame(frame)
+            if result == ProcessFrameResult.STOP:
+                break
 
         for strategy in self._user_turn_strategies.stop or []:
-            await strategy.process_frame(frame)
+            result = await strategy.process_frame(frame)
+            if result == ProcessFrameResult.STOP:
+                break
 
     async def _setup_strategies(self):
         for s in self._user_turn_strategies.start or []:
@@ -172,6 +181,7 @@ class UserTurnController(BaseObject):
             s.add_event_handler("on_push_frame", self._on_push_frame)
             s.add_event_handler("on_broadcast_frame", self._on_broadcast_frame)
             s.add_event_handler("on_user_turn_started", self._on_user_turn_started)
+            s.add_event_handler("on_reset_aggregation", self._on_reset_aggregation)
 
         for s in self._user_turn_strategies.stop or []:
             await s.setup(self.task_manager)
@@ -242,6 +252,9 @@ class UserTurnController(BaseObject):
     ):
         await self._trigger_user_turn_stop(strategy, params)
 
+    async def _on_reset_aggregation(self, strategy: BaseUserTurnStartStrategy):
+        await self._call_event_handler("on_reset_aggregation", strategy)
+
     async def _trigger_user_turn_start(
         self, strategy: Optional[BaseUserTurnStartStrategy], params: UserTurnStartedParams
     ):
@@ -256,6 +269,10 @@ class UserTurnController(BaseObject):
         for s in self._user_turn_strategies.start or []:
             await s.reset()
 
+        # Reset all user turn stop strategies to start fresh for the new turn.
+        for s in self._user_turn_strategies.stop or []:
+            await s.reset()
+
         await self._call_event_handler("on_user_turn_started", strategy, params)
 
     async def _trigger_user_turn_stop(

src/pipecat/turns/user_turn_strategies.py

@@ -23,6 +23,31 @@ from pipecat.turns.user_stop import (
 )
 
 
+def default_user_turn_start_strategies() -> List[BaseUserTurnStartStrategy]:
+    """Return the default user turn start strategies.
+
+    Returns ``[VADUserTurnStartStrategy, TranscriptionUserTurnStartStrategy]``.
+    Useful when building a custom strategy list that extends the defaults.
+
+    Example::
+
+        start_strategies = [
+            WakePhraseUserTurnStartStrategy(phrases=["hey pipecat"]),
+            *default_user_turn_start_strategies(),
+        ]
+    """
+    return [VADUserTurnStartStrategy(), TranscriptionUserTurnStartStrategy()]
+
+
+def default_user_turn_stop_strategies() -> List[BaseUserTurnStopStrategy]:
+    """Return the default user turn stop strategies.
+
+    Returns ``[TurnAnalyzerUserTurnStopStrategy(LocalSmartTurnAnalyzerV3)]``.
+    Useful when building a custom strategy list that extends the defaults.
+    """
+    return [TurnAnalyzerUserTurnStopStrategy(turn_analyzer=LocalSmartTurnAnalyzerV3())]
+
+
 @dataclass
 class UserTurnStrategies:
     """Container for user turn start and stop strategies.
@@ -45,9 +70,9 @@ class UserTurnStrategies:
 
     def __post_init__(self):
         if not self.start:
-            self.start = [VADUserTurnStartStrategy(), TranscriptionUserTurnStartStrategy()]
+            self.start = default_user_turn_start_strategies()
         if not self.stop:
-            self.stop = [TurnAnalyzerUserTurnStopStrategy(turn_analyzer=LocalSmartTurnAnalyzerV3())]
+            self.stop = default_user_turn_stop_strategies()
 
 
 @dataclass

src/pipecat/utils/tracing/service_attributes.py

@@ -25,20 +25,20 @@ if is_tracing_available():
     from opentelemetry.trace import Span
 
 
-def _get_gen_ai_system_from_service_name(service_name: str) -> str:
-    """Extract the standardized gen_ai.system value from a service class name.
+def _get_provider_name_from_service_name(service_name: str) -> str:
+    """Extract the standardized gen_ai.provider.name value from a service class name.
 
     Source:
-    https://opentelemetry.io/docs/specs/semconv/attributes-registry/gen-ai/#gen-ai-system
+    https://opentelemetry.io/docs/specs/semconv/attributes-registry/gen-ai/
 
     Uses standard OTel names where possible, with special case mappings for
     service names that don't follow the pattern.
 
     Args:
-        service_name: The service class name to extract system name from.
+        service_name: The service class name to extract provider name from.
 
     Returns:
-        The standardized gen_ai.system value.
+        The standardized gen_ai.provider.name value.
     """
     SPECIAL_CASE_MAPPINGS = {
         # AWS
@@ -91,7 +91,7 @@ def add_tts_span_attributes(
         **kwargs: Additional attributes to add.
     """
     # Add standard attributes
-    span.set_attribute("gen_ai.system", service_name.replace("TTSService", "").lower())
+    span.set_attribute("gen_ai.provider.name", service_name.replace("TTSService", "").lower())
     span.set_attribute("gen_ai.request.model", model)
     span.set_attribute("gen_ai.operation.name", operation_name)
     span.set_attribute("gen_ai.output.type", "speech")
@@ -150,7 +150,7 @@ def add_stt_span_attributes(
         **kwargs: Additional attributes to add.
     """
     # Add standard attributes
-    span.set_attribute("gen_ai.system", service_name.replace("STTService", "").lower())
+    span.set_attribute("gen_ai.provider.name", service_name.replace("STTService", "").lower())
     span.set_attribute("gen_ai.request.model", model)
     span.set_attribute("gen_ai.operation.name", operation_name)
     span.set_attribute("vad_enabled", vad_enabled)
@@ -193,7 +193,7 @@ def add_llm_span_attributes(
     tools: Optional[str] = None,
     tool_count: Optional[int] = None,
     tool_choice: Optional[str] = None,
-    system: Optional[str] = None,
+    system_instructions: Optional[str] = None,
     parameters: Optional[Dict[str, Any]] = None,
     extra_parameters: Optional[Dict[str, Any]] = None,
     ttfb: Optional[float] = None,
@@ -211,14 +211,14 @@ def add_llm_span_attributes(
         tools: JSON-serialized tools configuration.
         tool_count: Number of tools available.
         tool_choice: Tool selection configuration.
-        system: System message.
+        system_instructions: System instructions.
         parameters: Service parameters.
         extra_parameters: Additional parameters.
         ttfb: Time to first byte in seconds.
         **kwargs: Additional attributes to add.
     """
     # Add standard attributes
-    span.set_attribute("gen_ai.system", _get_gen_ai_system_from_service_name(service_name))
+    span.set_attribute("gen_ai.provider.name", _get_provider_name_from_service_name(service_name))
     span.set_attribute("gen_ai.request.model", model)
     span.set_attribute("gen_ai.operation.name", "chat")
     span.set_attribute("gen_ai.output.type", "text")
@@ -240,8 +240,8 @@ def add_llm_span_attributes(
     if tool_choice:
         span.set_attribute("tool_choice", tool_choice)
 
-    if system:
-        span.set_attribute("system", system)
+    if system_instructions:
+        span.set_attribute("gen_ai.system_instructions", system_instructions)
 
     if ttfb is not None:
         span.set_attribute("metrics.ttfb", ttfb)
@@ -313,7 +313,7 @@ def add_gemini_live_span_attributes(
         **kwargs: Additional attributes to add.
     """
     # Add standard attributes
-    span.set_attribute("gen_ai.system", "gcp.gemini")
+    span.set_attribute("gen_ai.provider.name", "gcp.gemini")
     span.set_attribute("gen_ai.request.model", model)
     span.set_attribute("gen_ai.operation.name", operation_name)
     span.set_attribute("service.operation", operation_name)
@@ -414,7 +414,7 @@ def add_openai_realtime_span_attributes(
         **kwargs: Additional attributes to add.
     """
     # Add standard attributes
-    span.set_attribute("gen_ai.system", "openai")
+    span.set_attribute("gen_ai.provider.name", "openai")
     span.set_attribute("gen_ai.request.model", model)
     span.set_attribute("gen_ai.operation.name", operation_name)
     span.set_attribute("service.operation", operation_name)

src/pipecat/utils/tracing/service_decorators.py

@@ -51,8 +51,10 @@ def _get_model_name(service) -> str:
     check all the places we used to store it.
     """
     return (
-        getattr(getattr(service, "_settings", None), "model", None)
-        or getattr(service, "_full_model_name", None)
+        # Some services store an API-response-provided detailed "full" name,
+        # which is distinct from the user-provided model name
+        getattr(service, "_full_model_name", None)
+        or getattr(getattr(service, "_settings", None), "model", None)
         or getattr(service, "model_name", None)
         or getattr(service, "_model_name", None)
         or "unknown"
@@ -135,14 +137,14 @@ def _add_token_usage_to_span(span, token_usage):
             and token_usage["cache_read_input_tokens"] is not None
         ):
             span.set_attribute(
-                "gen_ai.usage.cache_read_input_tokens", token_usage["cache_read_input_tokens"]
+                "gen_ai.usage.cache_read.input_tokens", token_usage["cache_read_input_tokens"]
             )
         if (
             "cache_creation_input_tokens" in token_usage
             and token_usage["cache_creation_input_tokens"] is not None
         ):
             span.set_attribute(
-                "gen_ai.usage.cache_creation_input_tokens",
+                "gen_ai.usage.cache_creation.input_tokens",
                 token_usage["cache_creation_input_tokens"],
             )
         if "reasoning_tokens" in token_usage and token_usage["reasoning_tokens"] is not None:
@@ -157,11 +159,11 @@ def _add_token_usage_to_span(span, token_usage):
         # Add cached token metrics for LLMTokenUsage object
         cache_read_tokens = getattr(token_usage, "cache_read_input_tokens", None)
         if cache_read_tokens is not None:
-            span.set_attribute("gen_ai.usage.cache_read_input_tokens", cache_read_tokens)
+            span.set_attribute("gen_ai.usage.cache_read.input_tokens", cache_read_tokens)
 
         cache_creation_tokens = getattr(token_usage, "cache_creation_input_tokens", None)
         if cache_creation_tokens is not None:
-            span.set_attribute("gen_ai.usage.cache_creation_input_tokens", cache_creation_tokens)
+            span.set_attribute("gen_ai.usage.cache_creation.input_tokens", cache_creation_tokens)
 
         reasoning_tokens = getattr(token_usage, "reasoning_tokens", None)
         if reasoning_tokens is not None:
@@ -500,18 +502,45 @@ def traced_llm(func: Optional[Callable] = None, *, name: Optional[str] = None) -
 
                             # Handle system message for different services
                             system_message = None
-                            if hasattr(context, "system"):
+                            if isinstance(context, LLMContext):
+                                # settings.system_instruction takes priority (matches service behavior)
+                                if hasattr(self, "_settings") and getattr(
+                                    self._settings, "system_instruction", None
+                                ):
+                                    system_message = self._settings.system_instruction
+                                else:
+                                    # Fall back to extracting from context messages
+                                    ctx_messages = context.get_messages()
+                                    if ctx_messages:
+                                        first = ctx_messages[0]
+                                        if (
+                                            isinstance(first, dict)
+                                            and first.get("role") == "system"
+                                        ):
+                                            content = first.get("content")
+                                            if isinstance(content, str):
+                                                system_message = content
+                                            elif isinstance(content, list):
+                                                system_message = " ".join(
+                                                    part.get("text", "")
+                                                    for part in content
+                                                    if isinstance(part, dict)
+                                                    and part.get("type") == "text"
+                                                )
+                            elif hasattr(context, "system"):
                                 system_message = context.system
                             elif hasattr(context, "system_message"):
                                 system_message = context.system_message
-                            elif hasattr(self, "_system_instruction"):
-                                system_message = self._system_instruction
 
                             # Use given_fields() defensively in case a service doesn't
                             # initialize all settings.
                             params = {}
                             if hasattr(self, "_settings"):
                                 for key, value in self._settings.given_fields().items():
+                                    # system_instruction is already captured as the
+                                    # "system_instructions" span attribute above.
+                                    if key == "system_instruction":
+                                        continue
                                     if isinstance(value, (int, float, bool, str)):
                                         params[key] = value
                                     elif value is None:
@@ -532,7 +561,7 @@ def traced_llm(func: Optional[Callable] = None, *, name: Optional[str] = None) -
                                 attribute_kwargs["tools"] = serialized_tools
                                 attribute_kwargs["tool_count"] = tool_count
                             if system_message:
-                                attribute_kwargs["system"] = system_message
+                                attribute_kwargs["system_instructions"] = system_message
 
                             # Add all gathered attributes to the span
                             add_llm_span_attributes(span=current_span, **attribute_kwargs)

tests/genesys/test_genesys_serializer.py

@@ -76,6 +76,7 @@ class TestGenesysAudioHookSerializer:
 
         assert msg["type"] == "pong"
         assert msg["id"] == serializer.session_id
+        assert msg["parameters"] == {}
 
     def test_create_closed_response(self):
         """Test creating a closed response message."""
@@ -86,7 +87,7 @@ class TestGenesysAudioHookSerializer:
 
         assert msg["type"] == "closed"
         assert serializer.is_open is False
-        assert "parameters" not in msg  # No parameters when no output_variables
+        assert msg["parameters"] == {}  # Empty parameters when no output_variables
 
     def test_create_closed_response_with_output_variables(self):
         """Test creating a closed response with custom output variables."""

tests/test_deepgram_stt.py

@@ -0,0 +1,51 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import io
+
+import pytest
+from loguru import logger
+
+from pipecat.services.deepgram.stt import _derive_deepgram_urls
+
+
+@pytest.mark.parametrize(
+    "base_url, expected_ws, expected_http",
+    [
+        # Secure schemes
+        ("wss://mydeepgram.com", "wss://mydeepgram.com", "https://mydeepgram.com"),
+        ("https://mydeepgram.com", "wss://mydeepgram.com", "https://mydeepgram.com"),
+        # Insecure schemes (air-gapped deployments)
+        ("ws://mydeepgram.com", "ws://mydeepgram.com", "http://mydeepgram.com"),
+        ("http://mydeepgram.com", "ws://mydeepgram.com", "http://mydeepgram.com"),
+        # Bare hostname defaults to secure
+        ("mydeepgram.com", "wss://mydeepgram.com", "https://mydeepgram.com"),
+        # With port
+        ("ws://localhost:8080", "ws://localhost:8080", "http://localhost:8080"),
+        ("wss://localhost:443", "wss://localhost:443", "https://localhost:443"),
+        ("localhost:8080", "wss://localhost:8080", "https://localhost:8080"),
+        # With path
+        ("wss://host/v1/listen", "wss://host/v1/listen", "https://host/v1/listen"),
+        ("http://host/v1/listen", "ws://host/v1/listen", "http://host/v1/listen"),
+    ],
+)
+def test_derive_deepgram_urls(base_url, expected_ws, expected_http):
+    ws_url, http_url = _derive_deepgram_urls(base_url)
+    assert ws_url == expected_ws
+    assert http_url == expected_http
+
+
+def test_derive_deepgram_urls_unknown_scheme_warns():
+    sink = io.StringIO()
+    handler_id = logger.add(sink, format="{message}")
+    try:
+        ws_url, http_url = _derive_deepgram_urls("ftp://mydeepgram.com")
+        # Falls back to secure
+        assert ws_url == "wss://mydeepgram.com"
+        assert http_url == "https://mydeepgram.com"
+        assert "Unrecognized scheme" in sink.getvalue()
+    finally:
+        logger.remove(handler_id)