Merge branch 'main' into groundingMetadata

src/pipecat/audio/turn/base_turn_analyzer.py

@@ -76,6 +76,16 @@ class BaseTurnAnalyzer(ABC):
         """
         pass
 
+    @property
+    @abstractmethod
+    def params(self):
+        """Get the current turn analyzer parameters.
+
+        Returns:
+            Current turn analyzer configuration parameters.
+        """
+        pass
+
     @abstractmethod
     def append_audio(self, buffer: bytes, is_speech: bool) -> EndOfTurnState:
         """Appends audio data for analysis.

src/pipecat/audio/turn/smart_turn/base_smart_turn.py

@@ -87,6 +87,15 @@ class BaseSmartTurn(BaseTurnAnalyzer):
         """
         return self._speech_triggered
 
+    @property
+    def params(self) -> SmartTurnParams:
+        """Get the current smart turn parameters.
+
+        Returns:
+            Current smart turn configuration parameters.
+        """
+        return self._params
+
     def append_audio(self, buffer: bytes, is_speech: bool) -> EndOfTurnState:
         """Append audio data for turn analysis.
 

src/pipecat/audio/turn/smart_turn/local_smart_turn_v2.py

@@ -0,0 +1,192 @@
+#
+# Copyright (c) 2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Local PyTorch turn analyzer for on-device ML inference using the smart-turn-v2 model.
+
+This module provides a smart turn analyzer that uses PyTorch models for
+local end-of-turn detection without requiring network connectivity.
+"""
+
+from typing import Any, Dict
+
+import numpy as np
+from loguru import logger
+
+from pipecat.audio.turn.smart_turn.base_smart_turn import BaseSmartTurn
+
+try:
+    import torch
+    import torch.nn.functional as F
+    from torch import nn
+    from transformers import (
+        Wav2Vec2Config,
+        Wav2Vec2Model,
+        Wav2Vec2PreTrainedModel,
+        Wav2Vec2Processor,
+    )
+except ModuleNotFoundError as e:
+    logger.error(f"Exception: {e}")
+    logger.error(
+        "In order to use LocalSmartTurnAnalyzerV2, you need to `pip install pipecat-ai[local-smart-turn]`."
+    )
+    raise Exception(f"Missing module: {e}")
+
+
+class LocalSmartTurnAnalyzerV2(BaseSmartTurn):
+    """Local turn analyzer using the smart-turn-v2 PyTorch model.
+
+    Provides end-of-turn detection using locally-stored PyTorch models,
+    enabling offline operation without network dependencies. Uses
+    Wav2Vec2 architecture for audio sequence classification.
+    """
+
+    def __init__(self, *, smart_turn_model_path: str, **kwargs):
+        """Initialize the local PyTorch smart-turn-v2 analyzer.
+
+        Args:
+            smart_turn_model_path: Path to directory containing the PyTorch model
+                and feature extractor files. If empty, uses default HuggingFace model.
+            **kwargs: Additional arguments passed to BaseSmartTurn.
+        """
+        super().__init__(**kwargs)
+
+        if not smart_turn_model_path:
+            # Define the path to the pretrained model on Hugging Face
+            smart_turn_model_path = "pipecat-ai/smart-turn-v2"
+
+        logger.debug("Loading Local Smart Turn v2 model...")
+        # Load the pretrained model for sequence classification
+        self._turn_model = _Wav2Vec2ForEndpointing.from_pretrained(smart_turn_model_path)
+        # Load the corresponding feature extractor for preprocessing audio
+        self._turn_processor = Wav2Vec2Processor.from_pretrained(smart_turn_model_path)
+        # Set device to GPU if available, else CPU
+        self._device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+        # Move model to selected device and set it to evaluation mode
+        self._turn_model = self._turn_model.to(self._device)
+        self._turn_model.eval()
+        logger.debug("Loaded Local Smart Turn v2")
+
+    async def _predict_endpoint(self, audio_array: np.ndarray) -> Dict[str, Any]:
+        """Predict end-of-turn using local PyTorch model."""
+        inputs = self._turn_processor(
+            audio_array,
+            sampling_rate=16000,
+            padding="max_length",
+            truncation=True,
+            max_length=16000 * 16,  # 16 seconds at 16kHz
+            return_attention_mask=True,
+            return_tensors="pt",
+        )
+
+        # Move inputs to device
+        inputs = {k: v.to(self._device) for k, v in inputs.items()}
+
+        # Run inference
+        with torch.no_grad():
+            outputs = self._turn_model(**inputs)
+
+            # The model returns sigmoid probabilities directly in the logits field
+            probability = outputs["logits"][0].item()
+
+            # Make prediction (1 for Complete, 0 for Incomplete)
+            prediction = 1 if probability > 0.5 else 0
+
+        return {
+            "prediction": prediction,
+            "probability": probability,
+        }
+
+
+class _Wav2Vec2ForEndpointing(Wav2Vec2PreTrainedModel):
+    def __init__(self, config: Wav2Vec2Config):
+        super().__init__(config)
+        self.wav2vec2 = Wav2Vec2Model(config)
+
+        self.pool_attention = nn.Sequential(
+            nn.Linear(config.hidden_size, 256), nn.Tanh(), nn.Linear(256, 1)
+        )
+
+        self.classifier = nn.Sequential(
+            nn.Linear(config.hidden_size, 256),
+            nn.LayerNorm(256),
+            nn.GELU(),
+            nn.Dropout(0.1),
+            nn.Linear(256, 64),
+            nn.GELU(),
+            nn.Linear(64, 1),
+        )
+
+        for module in self.classifier:
+            if isinstance(module, nn.Linear):
+                module.weight.data.normal_(mean=0.0, std=0.1)
+                if module.bias is not None:
+                    module.bias.data.zero_()
+
+        for module in self.pool_attention:
+            if isinstance(module, nn.Linear):
+                module.weight.data.normal_(mean=0.0, std=0.1)
+                if module.bias is not None:
+                    module.bias.data.zero_()
+
+    def attention_pool(self, hidden_states, attention_mask):
+        # Calculate attention weights
+        attention_weights = self.pool_attention(hidden_states)
+
+        if attention_mask is None:
+            raise ValueError("attention_mask must be provided for attention pooling")
+
+        attention_weights = attention_weights + (
+            (1.0 - attention_mask.unsqueeze(-1).to(attention_weights.dtype)) * -1e9
+        )
+
+        attention_weights = F.softmax(attention_weights, dim=1)
+
+        # Apply attention to hidden states
+        weighted_sum = torch.sum(hidden_states * attention_weights, dim=1)
+
+        return weighted_sum
+
+    def forward(self, input_values, attention_mask=None, labels=None):
+        outputs = self.wav2vec2(input_values, attention_mask=attention_mask)
+        hidden_states = outputs[0]
+
+        # Create transformer padding mask
+        if attention_mask is not None:
+            input_length = attention_mask.size(1)
+            hidden_length = hidden_states.size(1)
+            ratio = input_length / hidden_length
+            indices = (torch.arange(hidden_length, device=attention_mask.device) * ratio).long()
+            attention_mask = attention_mask[:, indices]
+            attention_mask = attention_mask.bool()
+        else:
+            attention_mask = None
+
+        pooled = self.attention_pool(hidden_states, attention_mask)
+
+        logits = self.classifier(pooled)
+
+        if torch.isnan(logits).any():
+            raise ValueError("NaN values detected in logits")
+
+        if labels is not None:
+            # Calculate positive sample weight based on batch statistics
+            pos_weight = ((labels == 0).sum() / (labels == 1).sum()).clamp(min=0.1, max=10.0)
+            loss_fct = nn.BCEWithLogitsLoss(pos_weight=pos_weight)
+            labels = labels.float()
+            loss = loss_fct(logits.view(-1), labels.view(-1))
+
+            # Add L2 regularization for classifier layers
+            l2_lambda = 0.01
+            l2_reg = torch.tensor(0.0, device=logits.device)
+            for param in self.classifier.parameters():
+                l2_reg += torch.norm(param)
+            loss += l2_lambda * l2_reg
+
+            probs = torch.sigmoid(logits.detach())
+            return {"loss": loss, "logits": probs}
+
+        probs = torch.sigmoid(logits)
+        return {"logits": probs}

src/pipecat/audio/vad/vad_analyzer.py

@@ -183,36 +183,37 @@ class VADAnalyzer(ABC):
         if len(self._vad_buffer) < num_required_bytes:
             return self._vad_state
 
-        audio_frames = self._vad_buffer[:num_required_bytes]
-        self._vad_buffer = self._vad_buffer[num_required_bytes:]
+        while len(self._vad_buffer) >= num_required_bytes:
+            audio_frames = self._vad_buffer[:num_required_bytes]
+            self._vad_buffer = self._vad_buffer[num_required_bytes:]
 
-        confidence = self.voice_confidence(audio_frames)
+            confidence = self.voice_confidence(audio_frames)
 
-        volume = self._get_smoothed_volume(audio_frames)
-        self._prev_volume = volume
+            volume = self._get_smoothed_volume(audio_frames)
+            self._prev_volume = volume
 
-        speaking = confidence >= self._params.confidence and volume >= self._params.min_volume
+            speaking = confidence >= self._params.confidence and volume >= self._params.min_volume
 
-        if speaking:
-            match self._vad_state:
-                case VADState.QUIET:
-                    self._vad_state = VADState.STARTING
-                    self._vad_starting_count = 1
-                case VADState.STARTING:
-                    self._vad_starting_count += 1
-                case VADState.STOPPING:
-                    self._vad_state = VADState.SPEAKING
-                    self._vad_stopping_count = 0
-        else:
-            match self._vad_state:
-                case VADState.STARTING:
-                    self._vad_state = VADState.QUIET
-                    self._vad_starting_count = 0
-                case VADState.SPEAKING:
-                    self._vad_state = VADState.STOPPING
-                    self._vad_stopping_count = 1
-                case VADState.STOPPING:
-                    self._vad_stopping_count += 1
+            if speaking:
+                match self._vad_state:
+                    case VADState.QUIET:
+                        self._vad_state = VADState.STARTING
+                        self._vad_starting_count = 1
+                    case VADState.STARTING:
+                        self._vad_starting_count += 1
+                    case VADState.STOPPING:
+                        self._vad_state = VADState.SPEAKING
+                        self._vad_stopping_count = 0
+            else:
+                match self._vad_state:
+                    case VADState.STARTING:
+                        self._vad_state = VADState.QUIET
+                        self._vad_starting_count = 0
+                    case VADState.SPEAKING:
+                        self._vad_state = VADState.STOPPING
+                        self._vad_stopping_count = 1
+                    case VADState.STOPPING:
+                        self._vad_stopping_count += 1
 
         if (
             self._vad_state == VADState.STARTING

src/pipecat/examples/run.py

@@ -9,6 +9,21 @@
 This module provides a unified interface for running Pipecat examples across
 different transport types including Daily.co, WebRTC, and Twilio. It handles
 setup, configuration, and lifecycle management for each transport type.
+
+Example usage:
+    SmallWebRTCTransport::
+
+        python bot.py --transport webrtc
+
+    DailyTransport::
+
+        python bot.py --transport daily
+
+    Twilio::
+
+        python bot.py --transport twilio --proxy username.ngrok.io
+        # Note: Concurrently, run an ngrok tunnel to your local server:
+        # ngrok http 7860
 """
 
 import argparse

src/pipecat/frames/frames.py

@@ -28,6 +28,7 @@ from typing import (
 )
 
 from pipecat.audio.interruptions.base_interruption_strategy import BaseInterruptionStrategy
+from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams
 from pipecat.audio.vad.vad_analyzer import VADParams
 from pipecat.metrics.metrics import MetricsData
 from pipecat.transcriptions.language import Language
@@ -1145,6 +1146,23 @@ class OutputDTMFUrgentFrame(DTMFFrame, SystemFrame):
     pass
 
 
+@dataclass
+class SpeechControlParamsFrame(SystemFrame):
+    """Frame for notifying processors of speech control parameter changes.
+
+    This includes parameters for both VAD (Voice Activity Detection) and
+    turn-taking analysis. It allows downstream processors to adjust their
+    behavior based on updated interaction control settings.
+
+    Parameters:
+        vad_params: Current VAD parameters.
+        turn_params: Current turn-taking analysis parameters.
+    """
+
+    vad_params: Optional[VADParams] = None
+    turn_params: Optional[SmartTurnParams] = None
+
+
 #
 # Control frames
 #

src/pipecat/pipeline/parallel_pipeline.py

@@ -273,12 +273,17 @@ class ParallelPipeline(BasePipeline):
         if not self._down_task:
             self._down_task = self.create_task(self._process_down_queue())
 
+    async def _drain_queue(self, queue: asyncio.Queue):
+        try:
+            while not queue.empty():
+                queue.get_nowait()
+        except asyncio.QueueEmpty:
+            logger.debug(f"Draining {self} queue already empty")
+
     async def _drain_queues(self):
         """Drain all frames from upstream and downstream queues."""
-        while not self._up_queue.empty:
-            await self._up_queue.get()
-        while not self._down_queue.empty:
-            await self._down_queue.get()
+        await self._drain_queue(self._up_queue)
+        await self._drain_queue(self._down_queue)
 
     async def _handle_interruption(self):
         """Handle interruption by cancelling tasks, draining queues, and restarting."""

src/pipecat/processors/aggregators/llm_response.py

@@ -19,6 +19,8 @@ from typing import Dict, List, Literal, Optional, Set
 from loguru import logger
 
 from pipecat.audio.interruptions.base_interruption_strategy import BaseInterruptionStrategy
+from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams
+from pipecat.audio.vad.vad_analyzer import VADParams
 from pipecat.frames.frames import (
     BotInterruptionFrame,
     BotStartedSpeakingFrame,
@@ -43,6 +45,7 @@ from pipecat.frames.frames import (
     LLMSetToolsFrame,
     LLMTextFrame,
     OpenAILLMContextAssistantTimestampFrame,
+    SpeechControlParamsFrame,
     StartFrame,
     StartInterruptionFrame,
     TextFrame,
@@ -67,9 +70,13 @@ class LLMUserAggregatorParams:
         aggregation_timeout: Maximum time in seconds to wait for additional
             transcription content before pushing aggregated result. This
             timeout is used only when the transcription is slow to arrive.
+        turn_emulated_vad_timeout: Maximum time in seconds to wait for emulated
+            VAD when using turn-based analysis. Applied when transcription is
+            received but VAD didn't detect speech (e.g., whispered utterances).
     """
 
     aggregation_timeout: float = 0.5
+    turn_emulated_vad_timeout: float = 0.8
 
 
 @dataclass
@@ -390,6 +397,9 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
         """
         super().__init__(context=context, role="user", **kwargs)
         self._params = params or LLMUserAggregatorParams()
+        self._vad_params: Optional[VADParams] = None
+        self._turn_params: Optional[SmartTurnParams] = None
+
         if "aggregation_timeout" in kwargs:
             import warnings
 
@@ -477,6 +487,10 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
             self.set_tools(frame.tools)
         elif isinstance(frame, LLMSetToolChoiceFrame):
             self.set_tool_choice(frame.tool_choice)
+        elif isinstance(frame, SpeechControlParamsFrame):
+            self._vad_params = frame.vad_params
+            self._turn_params = frame.turn_params
+            await self.push_frame(frame, direction)
         else:
             await self.push_frame(frame, direction)
 
@@ -618,9 +632,40 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
     async def _aggregation_task_handler(self):
         while True:
             try:
-                await asyncio.wait_for(
-                    self._aggregation_event.wait(), self._params.aggregation_timeout
-                )
+                # The _aggregation_task_handler handles two distinct timeout scenarios:
+                #
+                # 1. When emulating_vad=True: Wait for emulated VAD timeout before
+                #    pushing aggregation (simulating VAD behavior when no actual VAD
+                #    detection occurred).
+                #
+                # 2. When emulating_vad=False: Use aggregation_timeout as a buffer
+                #    to wait for potential late-arriving transcription frames after
+                #    a real VAD event.
+                #
+                # For emulated VAD scenarios, the timeout strategy depends on whether
+                # a turn analyzer is configured:
+                #
+                # - WITH turn analyzer: Use turn_emulated_vad_timeout parameter because
+                #   the VAD's stop_secs is set very low (e.g. 0.2s) for rapid speech
+                #   chunking to feed the turn analyzer. This low value is too fast
+                #   for emulated VAD scenarios where we need to allow users time to
+                #   finish speaking (e.g. 0.8s).
+                #
+                # - WITHOUT turn analyzer: Use VAD's stop_secs directly to maintain
+                #   consistent user experience between real VAD detection and
+                #   emulated VAD scenarios.
+                if not self._emulating_vad:
+                    timeout = self._params.aggregation_timeout
+                elif self._turn_params:
+                    timeout = self._params.turn_emulated_vad_timeout
+                else:
+                    # Use VAD stop_secs when no turn analyzer is present, fallback if no VAD params
+                    timeout = (
+                        self._vad_params.stop_secs
+                        if self._vad_params
+                        else self._params.turn_emulated_vad_timeout
+                    )
+                await asyncio.wait_for(self._aggregation_event.wait(), timeout)
                 await self._maybe_emulate_user_speaking()
             except asyncio.TimeoutError:
                 if not self._user_speaking:
@@ -648,7 +693,11 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
         # to emulate VAD (i.e. user start/stopped speaking), but we do it only
         # if the bot is not speaking. If the bot is speaking and we really have
         # a short utterance we don't really want to interrupt the bot.
-        if not self._user_speaking and not self._waiting_for_aggregation:
+        if (
+            not self._user_speaking
+            and not self._waiting_for_aggregation
+            and len(self._aggregation) > 0
+        ):
             if self._bot_speaking:
                 # If we reached this case and the bot is speaking, let's ignore
                 # what the user said.

src/pipecat/processors/frameworks/rtvi.py

@@ -44,6 +44,7 @@ from pipecat.frames.frames import (
     InterimTranscriptionFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
+    LLMMessagesAppendFrame,
     LLMTextFrame,
     MetricsFrame,
     StartFrame,
@@ -71,13 +72,14 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.services.llm_service import (
     FunctionCallParams,  # TODO(aleix): we shouldn't import `services` from `processors`
 )
+from pipecat.services.openai.llm import OpenAIContextAggregatorPair
 from pipecat.transports.base_input import BaseInputTransport
 from pipecat.transports.base_output import BaseOutputTransport
 from pipecat.transports.base_transport import BaseTransport
 from pipecat.utils.asyncio.watchdog_queue import WatchdogQueue
 from pipecat.utils.string import match_endofsentence
 
-RTVI_PROTOCOL_VERSION = "0.3.0"
+RTVI_PROTOCOL_VERSION = "1.0.0"
 
 RTVI_MESSAGE_LABEL = "rtvi-ai"
 RTVIMessageLiteral = Literal["rtvi-ai"]
@@ -90,6 +92,10 @@ class RTVIServiceOption(BaseModel):
 
     Defines a configurable option that can be set for an RTVI service,
     including its name, type, and handler function.
+
+    .. deprecated:: 0.0.75
+        Pipeline Configuration has been removed as part of the RTVI protocol 1.0.0.
+        Use custom client and server messages instead.
     """
 
     name: str
@@ -104,6 +110,10 @@ class RTVIService(BaseModel):
 
     Represents a service that can be configured and used within the RTVI protocol,
     containing a name and list of configurable options.
+
+    .. deprecated:: 0.0.75
+       Pipeline Configuration has been removed as part of the RTVI protocol 1.0.0.
+       Use custom client and server messages instead.
     """
 
     name: str
@@ -122,6 +132,10 @@ class RTVIActionArgumentData(BaseModel):
     """Data for an RTVI action argument.
 
     Contains the name and value of an argument passed to an RTVI action.
+
+    .. deprecated:: 0.0.75
+        Actions have been removed as part of the RTVI protocol 1.0.0.
+        Use custom client and server messages instead.
     """
 
     name: str
@@ -132,6 +146,10 @@ class RTVIActionArgument(BaseModel):
     """Definition of an RTVI action argument.
 
     Specifies the name and expected type of an argument for an RTVI action.
+
+    .. deprecated:: 0.0.75
+        Actions have been removed as part of the RTVI protocol 1.0.0.
+        Use custom client and server messages instead.
     """
 
     name: str
@@ -143,6 +161,10 @@ class RTVIAction(BaseModel):
 
     Represents an action that can be executed within the RTVI protocol,
     including its service, name, arguments, and handler function.
+
+    .. deprecated:: 0.0.75
+       Actions have been removed as part of the RTVI protocol 1.0.0.
+       Use custom client and server messages instead.
     """
 
     service: str
@@ -166,6 +188,10 @@ class RTVIServiceOptionConfig(BaseModel):
     """Configuration value for an RTVI service option.
 
     Contains the name and value to set for a specific service option.
+
+    .. deprecated:: 0.0.75
+        Pipeline Configuration has been removed as part of the RTVI protocol 1.0.0.
+        Use custom client and server messages instead.
     """
 
     name: str
@@ -176,6 +202,10 @@ class RTVIServiceConfig(BaseModel):
     """Configuration for an RTVI service.
 
     Contains the service name and list of option configurations to apply.
+
+    .. deprecated:: 0.0.75
+       Pipeline Configuration has been removed as part of the RTVI protocol 1.0.0.
+       Use custom client and server messages instead.
     """
 
     service: str
@@ -186,6 +216,10 @@ class RTVIConfig(BaseModel):
     """Complete RTVI configuration.
 
     Contains the full configuration for all RTVI services.
+
+    .. deprecated:: 0.0.75
+       Pipeline Configuration has been removed as part of the RTVI protocol 1.0.0.
+       Use custom client and server messages instead.
     """
 
     config: List[RTVIServiceConfig]
@@ -196,10 +230,15 @@ class RTVIConfig(BaseModel):
 #
 
 
+# deprecated
 class RTVIUpdateConfig(BaseModel):
     """Request to update RTVI configuration.
 
     Contains new configuration settings and whether to interrupt the bot.
+
+    .. deprecated:: 0.0.75
+        Pipeline Configuration has been removed as part of the RTVI protocol 1.0.0.
+        Use custom client and server messages instead.
     """
 
     config: List[RTVIServiceConfig]
@@ -210,6 +249,10 @@ class RTVIActionRunArgument(BaseModel):
     """Argument for running an RTVI action.
 
     Contains the name and value of an argument to pass to an action.
+
+    .. deprecated:: 0.0.75
+        Actions have been removed as part of the RTVI protocol 1.0.0.
+        Use custom client and server messages instead.
     """
 
     name: str
@@ -220,6 +263,10 @@ class RTVIActionRun(BaseModel):
     """Request to run an RTVI action.
 
     Contains the service, action name, and optional arguments.
+
+    .. deprecated:: 0.0.75
+       Actions have been removed as part of the RTVI protocol 1.0.0.
+       Use custom client and server messages instead.
     """
 
     service: str
@@ -234,12 +281,80 @@ class RTVIActionFrame(DataFrame):
     Parameters:
         rtvi_action_run: The action to execute.
         message_id: Optional message ID for response correlation.
+
+    .. deprecated:: 0.0.75
+        Actions have been removed as part of the RTVI protocol 1.0.0.
+        Use custom client and server messages instead.
     """
 
     rtvi_action_run: RTVIActionRun
     message_id: Optional[str] = None
 
 
+class RTVIRawClientMessageData(BaseModel):
+    """Data structure expected from client messages sent to the RTVI server."""
+
+    t: str
+    d: Optional[Any] = None
+
+
+class RTVIClientMessage(BaseModel):
+    """Cleansed data structure for client messages for handling."""
+
+    msg_id: str
+    type: str
+    data: Optional[Any] = None
+
+
+@dataclass
+class RTVIClientMessageFrame(SystemFrame):
+    """A frame for sending messages from the client to the RTVI server.
+
+    This frame is meant for custom messaging from the client to the server
+    and expects a server-response message.
+    """
+
+    msg_id: str
+    type: str
+    data: Optional[Any] = None
+
+
+@dataclass
+class RTVIServerResponseFrame(SystemFrame):
+    """A frame for responding to a client RTVI message.
+
+    This frame should be sent in response to an RTVIClientMessageFrame
+    and include the original RTVIClientMessageFrame to ensure the response
+    is properly attributed to the original request. To respond with an error,
+    set the `error` field to a string describing the error. This will result
+    in the client receiving a `response-error` message instead of a
+    `server-response` message.
+    """
+
+    client_msg: RTVIClientMessageFrame
+    data: Optional[Any] = None
+    error: Optional[str] = None
+
+
+class RTVIRawServerResponseData(BaseModel):
+    """Data structure for server responses to client messages."""
+
+    t: str
+    d: Optional[Any] = None
+
+
+class RTVIServerResponse(BaseModel):
+    """The RTVI-formatted message response from the server to the client.
+
+    This message is used to respond to custom messages sent by the client.
+    """
+
+    label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL
+    type: Literal["server-response"] = "server-response"
+    id: str
+    data: RTVIRawServerResponseData
+
+
 class RTVIMessage(BaseModel):
     """Base RTVI message structure.
 
@@ -269,7 +384,7 @@ class RTVIErrorResponseData(BaseModel):
 class RTVIErrorResponse(BaseModel):
     """RTVI error response message.
 
-    Sent in response to a client request that resulted in an error.
+    RTVI Formatted error response message for relaying failed client requests.
     """
 
     label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL
@@ -285,13 +400,13 @@ class RTVIErrorData(BaseModel):
     """
 
     error: str
-    fatal: bool
+    fatal: bool  # Indicates the pipeline has stopped due to this error
 
 
 class RTVIError(BaseModel):
     """RTVI error event message.
 
-    Sent when an error occurs that isn't in response to a specific request.
+    RTVI Formatted error message for relaying errors in the pipeline.
     """
 
     label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL
@@ -303,6 +418,10 @@ class RTVIDescribeConfigData(BaseModel):
     """Data for describing available RTVI configuration.
 
     Contains the list of available services and their options.
+
+    .. deprecated:: 0.0.75
+        Pipeline Configuration has been removed as part of the RTVI protocol 1.0.0.
+        Use custom client and server messages instead.
     """
 
     config: List[RTVIService]
@@ -312,6 +431,10 @@ class RTVIDescribeConfig(BaseModel):
     """Message describing available RTVI configuration.
 
     Sent in response to a describe-config request.
+
+    .. deprecated:: 0.0.75
+        Pipeline Configuration has been removed as part of the RTVI protocol 1.0.0.
+        Use custom client and server messages instead.
     """
 
     label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL
@@ -324,6 +447,10 @@ class RTVIDescribeActionsData(BaseModel):
     """Data for describing available RTVI actions.
 
     Contains the list of available actions that can be executed.
+
+    .. deprecated:: 0.0.75
+        Actions have been removed as part of the RTVI protocol 1.0.0.
+        Use custom client and server messages instead.
     """
 
     actions: List[RTVIAction]
@@ -333,6 +460,10 @@ class RTVIDescribeActions(BaseModel):
     """Message describing available RTVI actions.
 
     Sent in response to a describe-actions request.
+
+    .. deprecated:: 0.0.75
+        Actions have been removed as part of the RTVI protocol 1.0.0.
+        Use custom client and server messages instead.
     """
 
     label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL
@@ -345,6 +476,10 @@ class RTVIConfigResponse(BaseModel):
     """Response containing current RTVI configuration.
 
     Sent in response to a get-config request.
+
+    .. deprecated:: 0.0.75
+        Pipeline Configuration has been removed as part of the RTVI protocol 1.0.0.
+        Use custom client and server messages instead.
     """
 
     label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL
@@ -357,6 +492,10 @@ class RTVIActionResponseData(BaseModel):
     """Data for an RTVI action response.
 
     Contains the result of executing an action.
+
+    .. deprecated:: 0.0.75
+        Actions have been removed as part of the RTVI protocol 1.0.0.
+        Use custom client and server messages instead.
     """
 
     result: ActionResult
@@ -366,6 +505,10 @@ class RTVIActionResponse(BaseModel):
     """Response to an RTVI action execution.
 
     Sent after successfully executing an action.
+
+    .. deprecated:: 0.0.75
+        Actions have been removed as part of the RTVI protocol 1.0.0.
+        Use custom client and server messages instead.
     """
 
     label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL
@@ -374,6 +517,30 @@ class RTVIActionResponse(BaseModel):
     data: RTVIActionResponseData
 
 
+class AboutClientData(BaseModel):
+    """Data about the RTVI client.
+
+    Contains information about the client, including which RTVI library it
+    is using, what platform it is on and any additional details, if available.
+    """
+
+    library: str
+    library_version: Optional[str] = None
+    platform: Optional[str] = None
+    platform_version: Optional[str] = None
+    platform_details: Optional[Any] = None
+
+
+class RTVIClientReadyData(BaseModel):
+    """Data format of client ready messages.
+
+    Contains the RTVIprotocol version and client information.
+    """
+
+    version: str
+    about: AboutClientData
+
+
 class RTVIBotReadyData(BaseModel):
     """Data for bot ready notification.
 
@@ -381,7 +548,10 @@ class RTVIBotReadyData(BaseModel):
     """
 
     version: str
-    config: List[RTVIServiceConfig]
+    # The config field is deprecated and will not be included if
+    # the client's rtvi version is 1.0.0 or higher.
+    config: Optional[List[RTVIServiceConfig]] = None
+    about: Optional[Mapping[str, Any]] = None
 
 
 class RTVIBotReady(BaseModel):
@@ -418,6 +588,25 @@ class RTVILLMFunctionCallMessage(BaseModel):
     data: RTVILLMFunctionCallMessageData
 
 
+class RTVIAppendToContextData(BaseModel):
+    """Data format for appending messages to the context.
+
+    Contains the role, content, and whether to run the message immediately.
+    """
+
+    role: Literal["user", "assistant"] | str
+    content: Any
+    run_immediately: bool = False
+
+
+class RTVIAppendToContext(BaseModel):
+    """RTVI Message format to append content to the LLM context."""
+
+    label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL
+    type: Literal["append-to-context"] = "append-to-context"
+    data: RTVIAppendToContextData
+
+
 class RTVILLMFunctionCallStartMessageData(BaseModel):
     """Data for LLM function call start notification.
 
@@ -752,6 +941,11 @@ class RTVIObserver(BaseObserver):
         elif isinstance(frame, RTVIServerMessageFrame):
             message = RTVIServerMessage(data=frame.data)
             await self.push_transport_message_urgent(message)
+        elif isinstance(frame, RTVIServerResponseFrame):
+            if frame.error is not None:
+                await self._send_error_response(frame)
+            else:
+                await self._send_server_response(frame)
 
         if mark_as_seen:
             self._frames_seen.add(frame.id)
@@ -879,6 +1073,22 @@ class RTVIObserver(BaseObserver):
         message = RTVIMetricsMessage(data=metrics)
         await self.push_transport_message_urgent(message)
 
+    async def _send_server_response(self, frame: RTVIServerResponseFrame):
+        """Send a response to the client for a specific request."""
+        message = RTVIServerResponse(
+            id=str(frame.client_msg.msg_id),
+            data=RTVIRawServerResponseData(t=frame.client_msg.type, d=frame.data),
+        )
+        await self.push_transport_message_urgent(message)
+
+    async def _send_error_response(self, frame: RTVIServerResponseFrame):
+        """Send a response to the client for a specific request."""
+        if self._params.errors_enabled:
+            message = RTVIErrorResponse(
+                id=str(frame.client_msg.msg_id), data=RTVIErrorResponseData(error=frame.error)
+            )
+            await self.push_transport_message_urgent(message)
+
 
 class RTVIProcessor(FrameProcessor):
     """Main processor for handling RTVI protocol messages and actions.
@@ -908,6 +1118,7 @@ class RTVIProcessor(FrameProcessor):
         self._bot_ready = False
         self._client_ready = False
         self._client_ready_id = ""
+        self._client_version = []
         self._errors_enabled = True
 
         self._registered_actions: Dict[str, RTVIAction] = {}
@@ -921,6 +1132,7 @@ class RTVIProcessor(FrameProcessor):
 
         self._register_event_handler("on_bot_started")
         self._register_event_handler("on_client_ready")
+        self._register_event_handler("on_client_message")
 
         self._input_transport = None
         self._transport = transport
@@ -936,6 +1148,15 @@ class RTVIProcessor(FrameProcessor):
         Args:
             action: The action to register.
         """
+        import warnings
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "The actions API is deprecated, use server and client messages instead.",
+                DeprecationWarning,
+            )
+
         id = self._action_id(action.service, action.action)
         self._registered_actions[id] = action
 
@@ -945,6 +1166,15 @@ class RTVIProcessor(FrameProcessor):
         Args:
             service: The service to register.
         """
+        import warnings
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "The actions API is deprecated, use server and client messages instead.",
+                DeprecationWarning,
+            )
+
         self._registered_services[service.name] = service
 
     async def set_client_ready(self):
@@ -970,6 +1200,22 @@ class RTVIProcessor(FrameProcessor):
         """Send a bot interruption frame upstream."""
         await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM)
 
+    async def send_server_message(self, data: Any):
+        """Send a server message to the client."""
+        message = RTVIServerMessage(data=data)
+        await self._send_server_message(message)
+
+    async def send_server_response(self, client_msg: RTVIClientMessage, data: Any):
+        """Send a server response for a given client message."""
+        message = RTVIServerResponse(
+            id=client_msg.msg_id, data=RTVIRawServerResponseData(t=client_msg.type, d=data)
+        )
+        await self._send_server_message(message)
+
+    async def send_error_response(self, client_msg: RTVIClientMessage, error: str):
+        """Send an error response for a given client message."""
+        await self._send_error_response(id=client_msg.msg_id, error=error)
+
     async def send_error(self, error: str):
         """Send an error message to the client.
 
@@ -1013,9 +1259,6 @@ class RTVIProcessor(FrameProcessor):
             function_name: Name of the function being called.
             llm: The LLM processor making the call.
             context: The LLM context.
-
-        Note:
-            This method is deprecated. Use handle_function_call() instead.
         """
         import warnings
 
@@ -1136,7 +1379,15 @@ class RTVIProcessor(FrameProcessor):
         try:
             match message.type:
                 case "client-ready":
-                    await self._handle_client_ready(message.id)
+                    data = None
+                    try:
+                        data = RTVIClientReadyData.model_validate(message.data)
+                    except ValidationError:
+                        # Not all clients have been updated to RTVI 1.0.0.
+                        # For now, that's okay, we just log their info as unknown.
+                        data = None
+                        pass
+                    await self._handle_client_ready(message.id, data)
                 case "describe-actions":
                     await self._handle_describe_actions(message.id)
                 case "describe-config":
@@ -1148,6 +1399,9 @@ class RTVIProcessor(FrameProcessor):
                     await self._handle_update_config(message.id, update_config)
                 case "disconnect-bot":
                     await self.push_frame(EndTaskFrame(), FrameDirection.UPSTREAM)
+                case "client-message":
+                    data = RTVIRawClientMessageData.model_validate(message.data)
+                    await self._handle_client_message(message.id, data)
                 case "action":
                     action = RTVIActionRun.model_validate(message.data)
                     action_frame = RTVIActionFrame(message_id=message.id, rtvi_action_run=action)
@@ -1155,6 +1409,9 @@ class RTVIProcessor(FrameProcessor):
                 case "llm-function-call-result":
                     data = RTVILLMFunctionCallResultData.model_validate(message.data)
                     await self._handle_function_call_result(data)
+                case "append-to-context":
+                    data = RTVIAppendToContextData.model_validate(message.data)
+                    await self._handle_update_context(data)
                 case "raw-audio" | "raw-audio-batch":
                     await self._handle_audio_buffer(message.data)
 
@@ -1168,9 +1425,20 @@ class RTVIProcessor(FrameProcessor):
             await self._send_error_response(message.id, f"Exception processing message: {e}")
             logger.warning(f"Exception processing message: {e}")
 
-    async def _handle_client_ready(self, request_id: str):
-        """Handle a client-ready message."""
-        logger.debug("Received client-ready")
+    async def _handle_client_ready(self, request_id: str, data: RTVIClientReadyData | None):
+        """Handle the client-ready message from the client."""
+        version = data.version if data else "unknown"
+        logger.debug(f"Received client-ready: version {version}")
+        if version == "unknown":
+            self._client_version = [0, 3, 0]  # Default to 0.3.0 if unknown
+        else:
+            try:
+                self._client_version = [int(v) for v in version.split(".")]
+            except ValueError:
+                logger.warning(f"Invalid client version format: {version}")
+                self._client_version = [0, 3, 0]
+        about = data.about if data else {"library": "unknown"}
+        logger.debug(f"Client Details: {about}")
         if self._input_transport:
             await self._input_transport.start_audio_in_streaming()
 
@@ -1201,18 +1469,45 @@ class RTVIProcessor(FrameProcessor):
 
     async def _handle_describe_config(self, request_id: str):
         """Handle a describe-config request."""
+        import warnings
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "Configuration helpers are deprecated. If your application needs this behavior, use custom server and client messages.",
+                DeprecationWarning,
+            )
+
         services = list(self._registered_services.values())
         message = RTVIDescribeConfig(id=request_id, data=RTVIDescribeConfigData(config=services))
         await self._push_transport_message(message)
 
     async def _handle_describe_actions(self, request_id: str):
         """Handle a describe-actions request."""
+        import warnings
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "The Actions API is deprecated, use custom server and client messages instead.",
+                DeprecationWarning,
+            )
+
         actions = list(self._registered_actions.values())
         message = RTVIDescribeActions(id=request_id, data=RTVIDescribeActionsData(actions=actions))
         await self._push_transport_message(message)
 
     async def _handle_get_config(self, request_id: str):
         """Handle a get-config request."""
+        import warnings
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "Configuration helpers are deprecated. If your application needs this behavior, use custom server and client messages.",
+                DeprecationWarning,
+            )
+
         message = RTVIConfigResponse(id=request_id, data=self._config)
         await self._push_transport_message(message)
 
@@ -1230,6 +1525,15 @@ class RTVIProcessor(FrameProcessor):
 
     async def _update_service_config(self, config: RTVIServiceConfig):
         """Update configuration for a specific service."""
+        import warnings
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "Configuration helpers are deprecated. If your application needs this behavior, use custom server and client messages.",
+                DeprecationWarning,
+            )
+
         service = self._registered_services[config.service]
         for option in config.options:
             handler = service._options_dict[option.name].handler
@@ -1238,6 +1542,15 @@ class RTVIProcessor(FrameProcessor):
 
     async def _update_config(self, data: RTVIConfig, interrupt: bool):
         """Update the RTVI configuration."""
+        import warnings
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "Configuration helpers are deprecated. If your application needs this behavior, use custom server and client messages.",
+                DeprecationWarning,
+            )
+
         if interrupt:
             await self.interrupt_bot()
         for service_config in data.config:
@@ -1248,6 +1561,33 @@ class RTVIProcessor(FrameProcessor):
         await self._update_config(RTVIConfig(config=data.config), data.interrupt)
         await self._handle_get_config(request_id)
 
+    async def _handle_update_context(self, data: RTVIAppendToContextData):
+        if data.run_immediately:
+            await self.interrupt_bot()
+        frame = LLMMessagesAppendFrame(
+            messages=[{"role": data.role, "content": data.content}],
+            run_llm=data.run_immediately,
+        )
+        await self.push_frame(frame)
+
+    async def _handle_client_message(self, msg_id: str, data: RTVIRawClientMessageData):
+        """Handle a client message frame."""
+        if not data:
+            await self._send_error_response(msg_id, "Malformed client message")
+            return
+
+        # Create a RTVIClientMessageFrame to push the message
+        frame = RTVIClientMessageFrame(msg_id=msg_id, type=data.t, data=data.d)
+        await self.push_frame(frame)
+        await self._call_event_handler(
+            "on_client_message",
+            RTVIClientMessage(
+                msg_id=msg_id,
+                type=data.t,
+                data=data.d,
+            ),
+        )
+
     async def _handle_function_call_result(self, data):
         """Handle a function call result from the client."""
         frame = FunctionCallResultFrame(
@@ -1278,12 +1618,19 @@ class RTVIProcessor(FrameProcessor):
 
     async def _send_bot_ready(self):
         """Send the bot-ready message to the client."""
+        config = None
+        if self._client_version[0] < 1:
+            config = self._config.config
         message = RTVIBotReady(
             id=self._client_ready_id,
-            data=RTVIBotReadyData(version=RTVI_PROTOCOL_VERSION, config=self._config.config),
+            data=RTVIBotReadyData(version=RTVI_PROTOCOL_VERSION, config=config),
         )
         await self._push_transport_message(message)
 
+    async def _send_server_message(self, message: RTVIServerMessage | RTVIServerResponse):
+        """Send a message or response to the client."""
+        await self._push_transport_message(message)
+
     async def _send_error_frame(self, frame: ErrorFrame):
         """Send an error frame as an RTVI error message."""
         if self._errors_enabled:

src/pipecat/processors/user_idle_processor.py

@@ -15,6 +15,8 @@ from pipecat.frames.frames import (
     CancelFrame,
     EndFrame,
     Frame,
+    FunctionCallInProgressFrame,
+    FunctionCallResultFrame,
     StartFrame,
     UserStartedSpeakingFrame,
     UserStoppedSpeakingFrame,
@@ -168,6 +170,13 @@ class UserIdleProcessor(FrameProcessor):
                 self._idle_event.set()
             elif isinstance(frame, BotSpeakingFrame):
                 self._idle_event.set()
+            elif isinstance(frame, FunctionCallInProgressFrame):
+                # Function calls can take longer than the timeout, so we want to prevent idle callbacks
+                self._interrupted = True
+                self._idle_event.set()
+            elif isinstance(frame, FunctionCallResultFrame):
+                self._interrupted = False
+                self._idle_event.set()
 
     async def cleanup(self) -> None:
         """Cleans up resources when processor is shutting down."""

src/pipecat/serializers/exotel.py

@@ -108,6 +108,10 @@ class ExotelFrameSerializer(FrameSerializer):
             serialized_data = await self._output_resampler.resample(
                 data, frame.sample_rate, self._exotel_sample_rate
             )
+            if serialized_data is None or len(serialized_data) == 0:
+                # Ignoring in case we don't have audio
+                return None
+
             payload = base64.b64encode(serialized_data).decode("ascii")
 
             answer = {
@@ -144,6 +148,9 @@ class ExotelFrameSerializer(FrameSerializer):
                 self._exotel_sample_rate,
                 self._sample_rate,
             )
+            if deserialized_data is None or len(deserialized_data) == 0:
+                # Ignoring in case we don't have audio
+                return None
 
             # Input: Exotel takes PCM data, so just resample to match sample_rate
             audio_frame = InputAudioRawFrame(

src/pipecat/serializers/plivo.py

@@ -132,6 +132,10 @@ class PlivoFrameSerializer(FrameSerializer):
             serialized_data = await pcm_to_ulaw(
                 data, frame.sample_rate, self._plivo_sample_rate, self._output_resampler
             )
+            if serialized_data is None or len(serialized_data) == 0:
+                # Ignoring in case we don't have audio
+                return None
+
             payload = base64.b64encode(serialized_data).decode("utf-8")
             answer = {
                 "event": "playAudio",
@@ -227,6 +231,10 @@ class PlivoFrameSerializer(FrameSerializer):
             deserialized_data = await ulaw_to_pcm(
                 payload, self._plivo_sample_rate, self._sample_rate, self._input_resampler
             )
+            if deserialized_data is None or len(deserialized_data) == 0:
+                # Ignoring in case we don't have audio
+                return None
+
             audio_frame = InputAudioRawFrame(
                 audio=deserialized_data, num_channels=1, sample_rate=self._sample_rate
             )

src/pipecat/serializers/telnyx.py

@@ -155,6 +155,10 @@ class TelnyxFrameSerializer(FrameSerializer):
             else:
                 raise ValueError(f"Unsupported encoding: {self._params.inbound_encoding}")
 
+            if serialized_data is None or len(serialized_data) == 0:
+                # Ignoring in case we don't have audio
+                return None
+
             payload = base64.b64encode(serialized_data).decode("utf-8")
             answer = {
                 "event": "media",
@@ -262,6 +266,10 @@ class TelnyxFrameSerializer(FrameSerializer):
             else:
                 raise ValueError(f"Unsupported encoding: {self._params.outbound_encoding}")
 
+            if deserialized_data is None or len(deserialized_data) == 0:
+                # Ignoring in case we don't have audio
+                return None
+
             audio_frame = InputAudioRawFrame(
                 audio=deserialized_data, num_channels=1, sample_rate=self._sample_rate
             )

src/pipecat/serializers/twilio.py

@@ -132,6 +132,10 @@ class TwilioFrameSerializer(FrameSerializer):
             serialized_data = await pcm_to_ulaw(
                 data, frame.sample_rate, self._twilio_sample_rate, self._output_resampler
             )
+            if serialized_data is None or len(serialized_data) == 0:
+                # Ignoring in case we don't have audio
+                return None
+
             payload = base64.b64encode(serialized_data).decode("utf-8")
             answer = {
                 "event": "media",
@@ -185,8 +189,26 @@ class TwilioFrameSerializer(FrameSerializer):
                 async with session.post(endpoint, auth=auth, data=params) as response:
                     if response.status == 200:
                         logger.info(f"Successfully terminated Twilio call {call_sid}")
+                    elif response.status == 404:
+                        # Handle the case where the call has already ended
+                        # Error code 20404: "The requested resource was not found"
+                        # Source: https://www.twilio.com/docs/errors/20404
+                        try:
+                            error_data = await response.json()
+                            if error_data.get("code") == 20404:
+                                logger.debug(f"Twilio call {call_sid} was already terminated")
+                                return
+                        except:
+                            pass  # Fall through to log the raw error
+
+                        # Log other 404 errors
+                        error_text = await response.text()
+                        logger.error(
+                            f"Failed to terminate Twilio call {call_sid}: "
+                            f"Status {response.status}, Response: {error_text}"
+                        )
                     else:
-                        # Get the error details for better debugging
+                        # Log other errors
                         error_text = await response.text()
                         logger.error(
                             f"Failed to terminate Twilio call {call_sid}: "
@@ -217,6 +239,10 @@ class TwilioFrameSerializer(FrameSerializer):
             deserialized_data = await ulaw_to_pcm(
                 payload, self._twilio_sample_rate, self._sample_rate, self._input_resampler
             )
+            if deserialized_data is None or len(deserialized_data) == 0:
+                # Ignoring in case we don't have audio
+                return None
+
             audio_frame = InputAudioRawFrame(
                 audio=deserialized_data, num_channels=1, sample_rate=self._sample_rate
             )

src/pipecat/services/aws/llm.py

@@ -55,7 +55,7 @@ from pipecat.services.llm_service import LLMService
 from pipecat.utils.tracing.service_decorators import traced_llm
 
 try:
-    import boto3
+    import aioboto3
     import httpx
     from botocore.config import Config
 except ModuleNotFoundError as e:
@@ -749,13 +749,17 @@ class AWSBedrockLLMService(LLMService):
                 read_timeout=300,  # 5 minutes
                 retries={"max_attempts": 3},
             )
-        session = boto3.Session(
-            aws_access_key_id=aws_access_key,
-            aws_secret_access_key=aws_secret_key,
-            aws_session_token=aws_session_token,
-            region_name=aws_region,
-        )
-        self._client = session.client(service_name="bedrock-runtime", config=client_config)
+
+        self._aws_session = aioboto3.Session()
+
+        # Store AWS session parameters for creating client in async context
+        self._aws_params = {
+            "aws_access_key_id": aws_access_key,
+            "aws_secret_access_key": aws_secret_key,
+            "aws_session_token": aws_session_token,
+            "region_name": aws_region,
+            "config": client_config,
+        }
 
         self.set_model_name(model)
         self._settings = {
@@ -903,70 +907,74 @@ class AWSBedrockLLMService(LLMService):
 
             logger.debug(f"Calling AWS Bedrock model with: {request_params}")
 
-            # Call AWS Bedrock with streaming
-            response = self._client.converse_stream(**request_params)
+            async with self._aws_session.client(
+                service_name="bedrock-runtime", **self._aws_params
+            ) as client:
+                # Call AWS Bedrock with streaming
+                response = await client.converse_stream(**request_params)
 
-            await self.stop_ttfb_metrics()
+                await self.stop_ttfb_metrics()
 
-            # Process the streaming response
-            tool_use_block = None
-            json_accumulator = ""
+                # Process the streaming response
+                tool_use_block = None
+                json_accumulator = ""
 
-            function_calls = []
-            for event in response["stream"]:
-                self.reset_watchdog()
+                function_calls = []
 
-                # Handle text content
-                if "contentBlockDelta" in event:
-                    delta = event["contentBlockDelta"]["delta"]
-                    if "text" in delta:
-                        await self.push_frame(LLMTextFrame(delta["text"]))
-                        completion_tokens_estimate += self._estimate_tokens(delta["text"])
-                    elif "toolUse" in delta and "input" in delta["toolUse"]:
-                        # Handle partial JSON for tool use
-                        json_accumulator += delta["toolUse"]["input"]
-                        completion_tokens_estimate += self._estimate_tokens(
-                            delta["toolUse"]["input"]
-                        )
+                async for event in response["stream"]:
+                    self.reset_watchdog()
 
-                # Handle tool use start
-                elif "contentBlockStart" in event:
-                    content_block_start = event["contentBlockStart"]["start"]
-                    if "toolUse" in content_block_start:
-                        tool_use_block = {
-                            "id": content_block_start["toolUse"].get("toolUseId", ""),
-                            "name": content_block_start["toolUse"].get("name", ""),
-                        }
-                        json_accumulator = ""
+                    # Handle text content
+                    if "contentBlockDelta" in event:
+                        delta = event["contentBlockDelta"]["delta"]
+                        if "text" in delta:
+                            await self.push_frame(LLMTextFrame(delta["text"]))
+                            completion_tokens_estimate += self._estimate_tokens(delta["text"])
+                        elif "toolUse" in delta and "input" in delta["toolUse"]:
+                            # Handle partial JSON for tool use
+                            json_accumulator += delta["toolUse"]["input"]
+                            completion_tokens_estimate += self._estimate_tokens(
+                                delta["toolUse"]["input"]
+                            )
 
-                # Handle message completion with tool use
-                elif "messageStop" in event and "stopReason" in event["messageStop"]:
-                    if event["messageStop"]["stopReason"] == "tool_use" and tool_use_block:
-                        try:
-                            arguments = json.loads(json_accumulator) if json_accumulator else {}
+                    # Handle tool use start
+                    elif "contentBlockStart" in event:
+                        content_block_start = event["contentBlockStart"]["start"]
+                        if "toolUse" in content_block_start:
+                            tool_use_block = {
+                                "id": content_block_start["toolUse"].get("toolUseId", ""),
+                                "name": content_block_start["toolUse"].get("name", ""),
+                            }
+                            json_accumulator = ""
 
-                            # Only call function if it's not the no_operation tool
-                            if not using_noop_tool:
-                                function_calls.append(
-                                    FunctionCallFromLLM(
-                                        context=context,
-                                        tool_call_id=tool_use_block["id"],
-                                        function_name=tool_use_block["name"],
-                                        arguments=arguments,
+                    # Handle message completion with tool use
+                    elif "messageStop" in event and "stopReason" in event["messageStop"]:
+                        if event["messageStop"]["stopReason"] == "tool_use" and tool_use_block:
+                            try:
+                                arguments = json.loads(json_accumulator) if json_accumulator else {}
+
+                                # Only call function if it's not the no_operation tool
+                                if not using_noop_tool:
+                                    function_calls.append(
+                                        FunctionCallFromLLM(
+                                            context=context,
+                                            tool_call_id=tool_use_block["id"],
+                                            function_name=tool_use_block["name"],
+                                            arguments=arguments,
+                                        )
                                     )
-                                )
-                            else:
-                                logger.debug("Ignoring no_operation tool call")
-                        except json.JSONDecodeError:
-                            logger.error(f"Failed to parse tool arguments: {json_accumulator}")
+                                else:
+                                    logger.debug("Ignoring no_operation tool call")
+                            except json.JSONDecodeError:
+                                logger.error(f"Failed to parse tool arguments: {json_accumulator}")
 
-                # Handle usage metrics if available
-                if "metadata" in event and "usage" in event["metadata"]:
-                    usage = event["metadata"]["usage"]
-                    prompt_tokens += usage.get("inputTokens", 0)
-                    completion_tokens += usage.get("outputTokens", 0)
-                    cache_read_input_tokens += usage.get("cacheReadInputTokens", 0)
-                    cache_creation_input_tokens += usage.get("cacheWriteInputTokens", 0)
+                    # Handle usage metrics if available
+                    if "metadata" in event and "usage" in event["metadata"]:
+                        usage = event["metadata"]["usage"]
+                        prompt_tokens += usage.get("inputTokens", 0)
+                        completion_tokens += usage.get("outputTokens", 0)
+                        cache_read_input_tokens += usage.get("cacheReadInputTokens", 0)
+                        cache_creation_input_tokens += usage.get("cacheWriteInputTokens", 0)
 
             await self.run_function_calls(function_calls)
         except asyncio.CancelledError:

src/pipecat/services/aws/tts.py

@@ -30,7 +30,7 @@ from pipecat.transcriptions.language import Language
 from pipecat.utils.tracing.service_decorators import traced_tts
 
 try:
-    import boto3
+    import aioboto3
     from botocore.exceptions import BotoCoreError, ClientError
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
@@ -177,13 +177,25 @@ class AWSPollyTTSService(TTSService):
 
         params = params or AWSPollyTTSService.InputParams()
 
-        self._polly_client = boto3.client(
-            "polly",
-            aws_access_key_id=aws_access_key_id,
-            aws_secret_access_key=api_key,
-            aws_session_token=aws_session_token,
-            region_name=region,
-        )
+        # Get credentials from environment variables if not provided
+        self._aws_params = {
+            "aws_access_key_id": aws_access_key_id or os.getenv("AWS_ACCESS_KEY_ID"),
+            "aws_secret_access_key": api_key or os.getenv("AWS_SECRET_ACCESS_KEY"),
+            "aws_session_token": aws_session_token or os.getenv("AWS_SESSION_TOKEN"),
+            "region_name": region or os.getenv("AWS_REGION", "us-east-1"),
+        }
+
+        # Validate that we have the required credentials
+        if (
+            not self._aws_params["aws_access_key_id"]
+            or not self._aws_params["aws_secret_access_key"]
+        ):
+            raise ValueError(
+                "AWS credentials not found. Please provide them either through constructor parameters "
+                "or set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY environment variables."
+            )
+
+        self._aws_session = aioboto3.Session()
         self._settings = {
             "engine": params.engine,
             "language": self.language_to_service_language(params.language)
@@ -199,24 +211,6 @@ class AWSPollyTTSService(TTSService):
 
         self.set_voice(voice_id)
 
-        # Get credentials from environment variables if not provided
-        self._credentials = {
-            "aws_access_key_id": aws_access_key_id or os.getenv("AWS_ACCESS_KEY_ID"),
-            "aws_secret_access_key": api_key or os.getenv("AWS_SECRET_ACCESS_KEY"),
-            "aws_session_token": aws_session_token or os.getenv("AWS_SESSION_TOKEN"),
-            "region": region or os.getenv("AWS_REGION", "us-east-1"),
-        }
-
-        # Validate that we have the required credentials
-        if (
-            not self._credentials["aws_access_key_id"]
-            or not self._credentials["aws_secret_access_key"]
-        ):
-            raise ValueError(
-                "AWS credentials not found. Please provide them either through constructor parameters "
-                "or set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY environment variables."
-            )
-
     def can_generate_metrics(self) -> bool:
         """Check if this service can generate processing metrics.
 
@@ -279,14 +273,6 @@ class AWSPollyTTSService(TTSService):
         Yields:
             Frame: Audio frames containing the synthesized speech.
         """
-
-        def read_audio_data(**args):
-            response = self._polly_client.synthesize_speech(**args)
-            if "AudioStream" in response:
-                audio_data = response["AudioStream"].read()
-                return audio_data
-            return None
-
         logger.debug(f"{self}: Generating TTS [{text}]")
 
         try:
@@ -309,30 +295,32 @@ class AWSPollyTTSService(TTSService):
             # Filter out None values
             filtered_params = {k: v for k, v in params.items() if v is not None}
 
-            audio_data = await asyncio.to_thread(read_audio_data, **filtered_params)
+            async with self._aws_session.client("polly", **self._aws_params) as polly:
+                response = await polly.synthesize_speech(**filtered_params)
+                if "AudioStream" in response:
+                    # Get the streaming body and read it
+                    stream = response["AudioStream"]
+                    audio_data = await stream.read()
+                else:
+                    logger.error(f"{self} No audio stream in response")
+                    audio_data = None
 
-            if not audio_data:
-                logger.error(f"{self} No audio data returned")
-                yield None
-                return
+                audio_data = await self._resampler.resample(audio_data, 16000, self.sample_rate)
 
-            audio_data = await self._resampler.resample(audio_data, 16000, self.sample_rate)
+                await self.start_tts_usage_metrics(text)
 
-            await self.start_tts_usage_metrics(text)
+                yield TTSStartedFrame()
 
-            yield TTSStartedFrame()
+                CHUNK_SIZE = self.chunk_size
 
-            CHUNK_SIZE = self.chunk_size
-
-            for i in range(0, len(audio_data), CHUNK_SIZE):
-                chunk = audio_data[i : i + CHUNK_SIZE]
-                if len(chunk) > 0:
-                    await self.stop_ttfb_metrics()
-                    frame = TTSAudioRawFrame(chunk, self.sample_rate, 1)
-                    yield frame
-
-            yield TTSStoppedFrame()
+                for i in range(0, len(audio_data), CHUNK_SIZE):
+                    chunk = audio_data[i : i + CHUNK_SIZE]
+                    if len(chunk) > 0:
+                        await self.stop_ttfb_metrics()
+                        frame = TTSAudioRawFrame(chunk, self.sample_rate, 1)
+                        yield frame
 
+                yield TTSStoppedFrame()
         except (BotoCoreError, ClientError) as error:
             logger.exception(f"{self} error generating TTS: {error}")
             error_message = f"AWS Polly TTS error: {str(error)}"

src/pipecat/services/aws_nova_sonic/aws.py

@@ -474,7 +474,6 @@ class AWSNovaSonicLLMService(LLMService):
         # If we need to, send assistant response trigger (depends on self._connected_time)
         if self._triggering_assistant_response:
             await self._send_assistant_response_trigger()
-            self._triggering_assistant_response = False
 
     async def _disconnect(self):
         try:
@@ -1105,7 +1104,6 @@ class AWSNovaSonicLLMService(LLMService):
         # Send the trigger audio, if we're fully connected and set up
         if self._connected_time is not None:
             await self._send_assistant_response_trigger()
-            self._triggering_assistant_response = False
 
     async def _send_assistant_response_trigger(self):
         if (
@@ -1113,46 +1111,51 @@ class AWSNovaSonicLLMService(LLMService):
         ):  # should never happen
             return
 
-        logger.debug("Sending assistant response trigger...")
+        try:
+            logger.debug("Sending assistant response trigger...")
 
-        chunk_duration = 0.02  # what we might get from InputAudioRawFrame
-        chunk_size = int(
-            chunk_duration
-            * self._params.input_sample_rate
-            * self._params.input_channel_count
-            * (self._params.input_sample_size / 8)
-        )  # e.g. 0.02 seconds of 16-bit (2-byte) PCM mono audio at 16kHz is 640 bytes
+            chunk_duration = 0.02  # what we might get from InputAudioRawFrame
+            chunk_size = int(
+                chunk_duration
+                * self._params.input_sample_rate
+                * self._params.input_channel_count
+                * (self._params.input_sample_size / 8)
+            )  # e.g. 0.02 seconds of 16-bit (2-byte) PCM mono audio at 16kHz is 640 bytes
 
-        # Lead with a bit of blank audio, if needed.
-        # It seems like the LLM can't quite "hear" the first little bit of audio sent on a
-        # connection.
-        current_time = time.time()
-        max_blank_audio_duration = 0.5
-        blank_audio_duration = (
-            max_blank_audio_duration - (current_time - self._connected_time)
-            if self._connected_time is not None
-            and (current_time - self._connected_time) < max_blank_audio_duration
-            else None
-        )
-        if blank_audio_duration:
-            logger.debug(
-                f"Leading assistant response trigger with {blank_audio_duration}s of blank audio"
+            # Lead with a bit of blank audio, if needed.
+            # It seems like the LLM can't quite "hear" the first little bit of audio sent on a
+            # connection.
+            current_time = time.time()
+            max_blank_audio_duration = 0.5
+            blank_audio_duration = (
+                max_blank_audio_duration - (current_time - self._connected_time)
+                if self._connected_time is not None
+                and (current_time - self._connected_time) < max_blank_audio_duration
+                else None
             )
-            blank_audio_chunk = b"\x00" * chunk_size
-            num_chunks = int(blank_audio_duration / chunk_duration)
-            for _ in range(num_chunks):
-                await self._send_user_audio_event(blank_audio_chunk)
-                await asyncio.sleep(chunk_duration)
+            if blank_audio_duration:
+                logger.debug(
+                    f"Leading assistant response trigger with {blank_audio_duration}s of blank audio"
+                )
+                blank_audio_chunk = b"\x00" * chunk_size
+                num_chunks = int(blank_audio_duration / chunk_duration)
+                for _ in range(num_chunks):
+                    await self._send_user_audio_event(blank_audio_chunk)
+                    await asyncio.sleep(chunk_duration)
 
-        # Send trigger audio
-        # NOTE: this audio *will* be transcribed and eventually make it into the context. That's OK:
-        # if we ever need to seed this service again with context it would make sense to include it
-        # since the instruction (i.e. the "wait for the trigger" instruction) will be part of the
-        # context as well.
-        audio_chunks = [
-            self._assistant_response_trigger_audio[i : i + chunk_size]
-            for i in range(0, len(self._assistant_response_trigger_audio), chunk_size)
-        ]
-        for chunk in audio_chunks:
-            await self._send_user_audio_event(chunk)
-            await asyncio.sleep(chunk_duration)
+            # Send trigger audio
+            # NOTE: this audio *will* be transcribed and eventually make it into the context. That's OK:
+            # if we ever need to seed this service again with context it would make sense to include it
+            # since the instruction (i.e. the "wait for the trigger" instruction) will be part of the
+            # context as well.
+            audio_chunks = [
+                self._assistant_response_trigger_audio[i : i + chunk_size]
+                for i in range(0, len(self._assistant_response_trigger_audio), chunk_size)
+            ]
+            for chunk in audio_chunks:
+                await self._send_user_audio_event(chunk)
+                await asyncio.sleep(chunk_duration)
+        finally:
+            # We need to clean up in case sending the trigger was cancelled, e.g. in the case of a user interruption.
+            # (An asyncio.CancelledError would be raised in that case.)
+            self._triggering_assistant_response = False

src/pipecat/services/cartesia/tts.py

@@ -121,6 +121,7 @@ class CartesiaTTSService(AudioContextWordTTSService):
         container: str = "raw",
         params: Optional[InputParams] = None,
         text_aggregator: Optional[BaseTextAggregator] = None,
+        aggregate_sentences: Optional[bool] = True,
         **kwargs,
     ):
         """Initialize the Cartesia TTS service.
@@ -136,6 +137,7 @@ class CartesiaTTSService(AudioContextWordTTSService):
             container: Audio container format.
             params: Additional input parameters for voice customization.
             text_aggregator: Custom text aggregator for processing input text.
+            aggregate_sentences: Whether to aggregate sentences within the TTSService.
             **kwargs: Additional arguments passed to the parent service.
         """
         # Aggregating sentences still gives cleaner-sounding results and fewer
@@ -149,7 +151,7 @@ class CartesiaTTSService(AudioContextWordTTSService):
         # can use those to generate text frames ourselves aligned with the
         # playout timing of the audio!
         super().__init__(
-            aggregate_sentences=True,
+            aggregate_sentences=aggregate_sentences,
             push_text_frames=False,
             pause_frame_processing=True,
             sample_rate=sample_rate,

src/pipecat/services/elevenlabs/tts.py

@@ -238,6 +238,7 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
         url: str = "wss://api.elevenlabs.io",
         sample_rate: Optional[int] = None,
         params: Optional[InputParams] = None,
+        aggregate_sentences: Optional[bool] = True,
         **kwargs,
     ):
         """Initialize the ElevenLabs TTS service.
@@ -249,6 +250,7 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
             url: WebSocket URL for ElevenLabs TTS API.
             sample_rate: Audio sample rate. If None, uses default.
             params: Additional input parameters for voice customization.
+            aggregate_sentences: Whether to aggregate sentences within the TTSService.
             **kwargs: Additional arguments passed to the parent service.
         """
         # Aggregating sentences still gives cleaner-sounding results and fewer
@@ -266,7 +268,7 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
         # speaking for a while, so we want the parent class to send TTSStopFrame
         # after a short period not receiving any audio.
         super().__init__(
-            aggregate_sentences=True,
+            aggregate_sentences=aggregate_sentences,
             push_text_frames=False,
             push_stop_frames=True,
             pause_frame_processing=True,

src/pipecat/services/google/llm.py

@@ -627,9 +627,9 @@ class GoogleLLMContext(OpenAILLMContext):
         # Check if we only have function-related messages (no regular text)
         has_regular_messages = any(
             len(msg.parts) == 1
-            and not getattr(msg.parts[0], "text", None)
-            and getattr(msg.parts[0], "function_call", None)
-            and getattr(msg.parts[0], "function_response", None)
+            and getattr(msg.parts[0], "text", None)
+            and not getattr(msg.parts[0], "function_call", None)
+            and not getattr(msg.parts[0], "function_response", None)
             for msg in self._messages
         )
 

src/pipecat/services/lmnt/tts.py

@@ -95,7 +95,7 @@ class LmntTTSService(InterruptibleTTSService):
         voice_id: str,
         sample_rate: Optional[int] = None,
         language: Language = Language.EN,
-        model: str = "aurora",
+        model: str = "blizzard",
         **kwargs,
     ):
         """Initialize the LMNT TTS service.
@@ -105,7 +105,7 @@ class LmntTTSService(InterruptibleTTSService):
             voice_id: ID of the voice to use for synthesis.
             sample_rate: Audio sample rate. If None, uses default.
             language: Language for synthesis. Defaults to English.
-            model: TTS model to use. Defaults to "aurora".
+            model: TTS model to use. Defaults to "blizzard".
             **kwargs: Additional arguments passed to parent InterruptibleTTSService.
         """
         super().__init__(

src/pipecat/services/mem0/memory.py

@@ -69,6 +69,7 @@ class Mem0MemoryService(FrameProcessor):
         agent_id: Optional[str] = None,
         run_id: Optional[str] = None,
         params: Optional[InputParams] = None,
+        host: Optional[str] = None,
     ):
         """Initialize the Mem0 memory service.
 
@@ -79,6 +80,7 @@ class Mem0MemoryService(FrameProcessor):
             agent_id: The agent ID to associate with memories in Mem0.
             run_id: The run ID to associate with memories in Mem0.
             params: Configuration parameters for memory retrieval and storage.
+            host: The host of the Mem0 server.
 
         Raises:
             ValueError: If none of user_id, agent_id, or run_id are provided.
@@ -92,7 +94,7 @@ class Mem0MemoryService(FrameProcessor):
         if local_config:
             self.memory_client = Memory.from_config(local_config)
         else:
-            self.memory_client = MemoryClient(api_key=api_key)
+            self.memory_client = MemoryClient(api_key=api_key, host=host)
         # At least one of user_id, agent_id, or run_id must be provided
         if not any([user_id, agent_id, run_id]):
             raise ValueError("At least one of user_id, agent_id, or run_id must be provided")

src/pipecat/services/neuphonic/tts.py

@@ -106,10 +106,11 @@ class NeuphonicTTSService(InterruptibleTTSService):
         *,
         api_key: str,
         voice_id: Optional[str] = None,
-        url: str = "wss://api.neuphonic.com",
+        url: str = "wss://eu-west-1.api.neuphonic.com",
         sample_rate: Optional[int] = 22050,
         encoding: str = "pcm_linear",
         params: Optional[InputParams] = None,
+        aggregate_sentences: Optional[bool] = True,
         **kwargs,
     ):
         """Initialize the Neuphonic TTS service.
@@ -121,10 +122,11 @@ class NeuphonicTTSService(InterruptibleTTSService):
             sample_rate: Audio sample rate in Hz. Defaults to 22050.
             encoding: Audio encoding format. Defaults to "pcm_linear".
             params: Additional input parameters for TTS configuration.
+            aggregate_sentences: Whether to aggregate sentences within the TTSService.
             **kwargs: Additional arguments passed to parent InterruptibleTTSService.
         """
         super().__init__(
-            aggregate_sentences=True,
+            aggregate_sentences=aggregate_sentences,
             push_text_frames=False,
             push_stop_frames=True,
             stop_frame_timeout_s=2.0,
@@ -279,14 +281,18 @@ class NeuphonicTTSService(InterruptibleTTSService):
                 "voice_id": self._voice_id,
             }
 
-            query_params = [f"api_key={self._api_key}"]
+            query_params = []
             for key, value in tts_config.items():
                 if value is not None:
                     query_params.append(f"{key}={value}")
 
-            url = f"{self._url}/speak/{self._settings['lang_code']}?{'&'.join(query_params)}"
+            url = f"{self._url}/speak/{self._settings['lang_code']}"
+            if query_params:
+                url += f"?{'&'.join(query_params)}"
 
-            self._websocket = await websockets.connect(url)
+            headers = {"x-api-key": self._api_key}
+
+            self._websocket = await websockets.connect(url, extra_headers=headers)
         except Exception as e:
             logger.error(f"{self} initialization error: {e}")
             self._websocket = None
@@ -311,7 +317,7 @@ class NeuphonicTTSService(InterruptibleTTSService):
         async for message in WatchdogAsyncIterator(self._websocket, manager=self.task_manager):
             if isinstance(message, str):
                 msg = json.loads(message)
-                if msg.get("data", {}).get("audio") is not None:
+                if msg.get("data") and msg["data"].get("audio"):
                     await self.stop_ttfb_metrics()
 
                     audio = base64.b64decode(msg["data"]["audio"])
@@ -324,12 +330,19 @@ class NeuphonicTTSService(InterruptibleTTSService):
         while True:
             self.reset_watchdog()
             await asyncio.sleep(KEEPALIVE_SLEEP)
-            await self._send_text("")
+            await self._send_keepalive()
+
+    async def _send_keepalive(self):
+        """Send keepalive message to maintain connection."""
+        if self._websocket:
+            # Send empty text for keepalive
+            msg = {"text": ""}
+            await self._websocket.send(json.dumps(msg))
 
     async def _send_text(self, text: str):
         """Send text to Neuphonic WebSocket for synthesis."""
         if self._websocket:
-            msg = {"text": text}
+            msg = {"text": f"{text} <STOP>"}
             logger.debug(f"Sending text to websocket: {msg}")
             await self._websocket.send(json.dumps(msg))
 

src/pipecat/services/ollama/llm.py

@@ -6,6 +6,8 @@
 
 """OLLama LLM service implementation for Pipecat AI framework."""
 
+from loguru import logger
+
 from pipecat.services.openai.llm import OpenAILLMService
 
 
@@ -16,12 +18,28 @@ class OLLamaLLMService(OpenAILLMService):
     providing a compatible interface for running large language models locally.
     """
 
-    def __init__(self, *, model: str = "llama2", base_url: str = "http://localhost:11434/v1"):
+    def __init__(
+        self, *, model: str = "llama2", base_url: str = "http://localhost:11434/v1", **kwargs
+    ):
         """Initialize OLLama LLM service.
 
         Args:
             model: The OLLama model to use. Defaults to "llama2".
             base_url: The base URL for the OLLama API endpoint.
                     Defaults to "http://localhost:11434/v1".
+            **kwargs: Additional keyword arguments passed to OpenAILLMService.
         """
-        super().__init__(model=model, base_url=base_url, api_key="ollama")
+        super().__init__(model=model, base_url=base_url, api_key="ollama", **kwargs)
+
+    def create_client(self, base_url=None, **kwargs):
+        """Create OpenAI-compatible client for Ollama.
+
+        Args:
+            base_url: The base URL for the API. If None, uses instance base_url.
+            **kwargs: Additional keyword arguments passed to the parent create_client method.
+
+        Returns:
+            An OpenAI-compatible client configured for Ollama.
+        """
+        logger.debug(f"Creating Ollama client with api {base_url}")
+        return super().create_client(base_url, **kwargs)

src/pipecat/services/rime/tts.py

@@ -99,6 +99,7 @@ class RimeTTSService(AudioContextWordTTSService):
         sample_rate: Optional[int] = None,
         params: Optional[InputParams] = None,
         text_aggregator: Optional[BaseTextAggregator] = None,
+        aggregate_sentences: Optional[bool] = True,
         **kwargs,
     ):
         """Initialize Rime TTS service.
@@ -111,11 +112,12 @@ class RimeTTSService(AudioContextWordTTSService):
             sample_rate: Audio sample rate in Hz.
             params: Additional configuration parameters.
             text_aggregator: Custom text aggregator for processing input text.
+            aggregate_sentences: Whether to aggregate sentences within the TTSService.
             **kwargs: Additional arguments passed to parent class.
         """
         # Initialize with parent class settings for proper frame handling
         super().__init__(
-            aggregate_sentences=True,
+            aggregate_sentences=aggregate_sentences,
             push_text_frames=False,
             push_stop_frames=True,
             pause_frame_processing=True,

src/pipecat/services/riva/stt.py

@@ -279,7 +279,6 @@ class RivaSTTService(STTService):
             streaming_config=self._config,
         )
         for response in responses:
-            self.reset_watchdog()
             if not response.results:
                 continue
             asyncio.run_coroutine_threadsafe(

src/pipecat/services/soniox/stt.py

@@ -0,0 +1,396 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Soniox speech-to-text service implementation."""
+
+import asyncio
+import json
+import time
+from typing import AsyncGenerator, List, Optional
+
+from loguru import logger
+from pydantic import BaseModel
+
+from pipecat.frames.frames import (
+    CancelFrame,
+    EndFrame,
+    ErrorFrame,
+    Frame,
+    InterimTranscriptionFrame,
+    StartFrame,
+    TranscriptionFrame,
+    UserStoppedSpeakingFrame,
+)
+from pipecat.processors.frame_processor import FrameDirection
+from pipecat.services.stt_service import STTService
+from pipecat.transcriptions.language import Language
+from pipecat.utils.time import time_now_iso8601
+from pipecat.utils.tracing.service_decorators import traced_stt
+
+try:
+    import websockets
+except ModuleNotFoundError as e:
+    logger.error(f"Exception: {e}")
+    logger.error("In order to use Soniox, you need to `pip install pipecat-ai[soniox]`.")
+    raise Exception(f"Missing module: {e}")
+
+
+KEEPALIVE_MESSAGE = '{"type": "keepalive"}'
+
+FINALIZE_MESSAGE = '{"type": "finalize"}'
+
+END_TOKEN = "<end>"
+
+FINALIZED_TOKEN = "<fin>"
+
+
+class SonioxInputParams(BaseModel):
+    """Real-time transcription settings.
+
+    See Soniox WebSocket API documentation for more details:
+    https://soniox.com/docs/speech-to-text/api-reference/websocket-api#configuration-parameters
+
+    Parameters:
+        model: Model to use for transcription.
+        audio_format: Audio format to use for transcription.
+        num_channels: Number of channels to use for transcription.
+        language_hints: List of language hints to use for transcription.
+        context: Customization for transcription.
+        enable_non_final_tokens: Whether to enable non-final tokens. If false, only final tokens will be returned.
+        max_non_final_tokens_duration_ms: Maximum duration of non-final tokens.
+        client_reference_id: Client reference ID to use for transcription.
+    """
+
+    model: str = "stt-rt-preview"
+
+    audio_format: Optional[str] = "pcm_s16le"
+    num_channels: Optional[int] = 1
+
+    language_hints: Optional[List[Language]] = None
+    context: Optional[str] = None
+
+    enable_non_final_tokens: Optional[bool] = True
+    max_non_final_tokens_duration_ms: Optional[int] = None
+
+    client_reference_id: Optional[str] = None
+
+
+def is_end_token(token: dict) -> bool:
+    """Determine if a token is an end token."""
+    return token["text"] == END_TOKEN or token["text"] == FINALIZED_TOKEN
+
+
+def language_to_soniox_language(language: Language) -> str:
+    """Pipecat Language enum uses same ISO 2-letter codes as Soniox, except with added regional variants.
+
+    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
+
+
+def _prepare_language_hints(
+    language_hints: Optional[List[Language]],
+) -> Optional[List[str]]:
+    if language_hints is None:
+        return None
+
+    prepared_languages = [language_to_soniox_language(lang) for lang in language_hints]
+    # Remove duplicates (in case of language_hints with multiple regions).
+    return list(set(prepared_languages))
+
+
+class SonioxSTTService(STTService):
+    """Speech-to-Text service using Soniox's WebSocket API.
+
+    This service connects to Soniox's WebSocket API for real-time transcription
+    with support for multiple languages, custom context, speaker diarization,
+    and more.
+
+    For complete API documentation, see: https://soniox.com/docs/speech-to-text/api-reference/websocket-api
+    """
+
+    def __init__(
+        self,
+        *,
+        api_key: str,
+        url: str = "wss://stt-rt.soniox.com/transcribe-websocket",
+        sample_rate: Optional[int] = None,
+        params: Optional[SonioxInputParams] = None,
+        vad_force_turn_endpoint: bool = False,
+        **kwargs,
+    ):
+        """Initialize the Soniox STT service.
+
+        Args:
+            api_key: Soniox API key.
+            url: Soniox WebSocket API URL.
+            sample_rate: Audio sample rate.
+            params: Additional configuration parameters, such as language hints, context and
+                speaker diarization.
+            vad_force_turn_endpoint: Listen to `UserStoppedSpeakingFrame` to send finalize message to Soniox. If disabled, Soniox will detect the end of the speech.
+            **kwargs: Additional arguments passed to the STTService.
+        """
+        super().__init__(sample_rate=sample_rate, **kwargs)
+        params = params or SonioxInputParams()
+
+        self._api_key = api_key
+        self._url = url
+        self.set_model_name(params.model)
+        self._params = params
+        self._vad_force_turn_endpoint = vad_force_turn_endpoint
+        self._websocket = None
+
+        self._final_transcription_buffer = []
+        self._last_tokens_received: Optional[float] = None
+
+        self._receive_task = None
+        self._keepalive_task = None
+
+    async def start(self, frame: StartFrame):
+        """Start the Soniox STT websocket connection.
+
+        Args:
+            frame: The start frame containing initialization parameters.
+        """
+        await super().start(frame)
+        if self._websocket:
+            return
+
+        self._websocket = await websockets.connect(self._url)
+
+        if not self._websocket:
+            logger.error(f"Unable to connect to Soniox API at {self._url}")
+
+        # If vad_force_turn_endpoint is not enabled, we need to enable endpoint detection.
+        # Either one or the other is required.
+        enable_endpoint_detection = not self._vad_force_turn_endpoint
+
+        # Send the initial configuration message.
+        config = {
+            "api_key": self._api_key,
+            "model": self._model_name,
+            "audio_format": self._params.audio_format,
+            "num_channels": self._params.num_channels or 1,
+            "enable_endpoint_detection": enable_endpoint_detection,
+            "sample_rate": self.sample_rate,
+            "language_hints": _prepare_language_hints(self._params.language_hints),
+            "context": self._params.context,
+            "enable_non_final_tokens": self._params.enable_non_final_tokens,
+            "max_non_final_tokens_duration_ms": self._params.max_non_final_tokens_duration_ms,
+            "client_reference_id": self._params.client_reference_id,
+        }
+
+        # Send the configuration message.
+        await self._websocket.send(json.dumps(config))
+
+        if self._websocket and not self._receive_task:
+            self._receive_task = self.create_task(self._receive_task_handler())
+        if self._websocket and not self._keepalive_task:
+            self._keepalive_task = self.create_task(self._keepalive_task_handler())
+
+    async def _cleanup(self):
+        if self._keepalive_task:
+            await self.cancel_task(self._keepalive_task)
+            self._keepalive_task = None
+
+        if self._websocket:
+            await self._websocket.close()
+            self._websocket = None
+
+        if self._receive_task:
+            # Task cannot cancel itself. If task called _cleanup() we expect it to cancel itself.
+            if self._receive_task != asyncio.current_task():
+                await self.wait_for_task(self._receive_task)
+            self._receive_task = None
+
+    async def stop(self, frame: EndFrame):
+        """Stop the Soniox STT websocket connection.
+
+        Stopping waits for the server to close the connection as we might receive
+        additional final tokens after sending the stop recording message.
+
+        Args:
+            frame: The end frame.
+        """
+        await super().stop(frame)
+        await self._send_stop_recording()
+
+    async def cancel(self, frame: CancelFrame):
+        """Cancel the Soniox STT websocket connection.
+
+        Compared to stop, this method closes the connection immediately without waiting
+        for the server to close it. This is useful when we want to stop the connection
+        immediately without waiting for the server to send any final tokens.
+
+        Args:
+            frame: The cancel frame.
+        """
+        await super().cancel(frame)
+        await self._cleanup()
+
+    async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
+        """Send audio data to Soniox STT Service.
+
+        Args:
+            audio: Raw audio bytes to transcribe.
+
+        Yields:
+            Frame: None (transcription results come via WebSocket callbacks).
+        """
+        await self.start_processing_metrics()
+        if self._websocket and not self._websocket.closed:
+            await self._websocket.send(audio)
+        await self.stop_processing_metrics()
+
+        yield None
+
+    @traced_stt
+    async def _handle_transcription(
+        self, transcript: str, is_final: bool, language: Optional[Language] = None
+    ):
+        """Handle a transcription result with tracing."""
+        pass
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        """Processes a frame of audio data, either buffering or transcribing it.
+
+        Args:
+            frame: The frame to process.
+            direction: The direction of frame processing.
+        """
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, UserStoppedSpeakingFrame) and self._vad_force_turn_endpoint:
+            # Send finalize message to Soniox so we get the final tokens asap.
+            if self._websocket and not self._websocket.closed:
+                await self._websocket.send(FINALIZE_MESSAGE)
+                logger.debug(f"Triggered finalize event on: {frame.name=}, {direction=}")
+
+    async def _send_stop_recording(self):
+        """Send stop recording message to Soniox."""
+        if self._websocket and not self._websocket.closed:
+            # Send stop recording message
+            await self._websocket.send("")
+
+    async def _keepalive_task_handler(self):
+        """Connection has to be open all the time."""
+        try:
+            while True:
+                logger.debug("Sending keepalive message")
+                if self._websocket and not self._websocket.closed:
+                    await self._websocket.send(KEEPALIVE_MESSAGE)
+                else:
+                    logger.debug("WebSocket connection closed.")
+                    break
+                await asyncio.sleep(5)
+
+        except websockets.exceptions.ConnectionClosed:
+            # Expected when closing the connection
+            logger.debug("WebSocket connection closed, keepalive task stopped.")
+        except Exception as e:
+            logger.error(f"{self} error (_keepalive_task_handler): {e}")
+            await self.push_error(ErrorFrame(f"{self} error (_keepalive_task_handler): {e}"))
+
+    async def _receive_task_handler(self):
+        if not self._websocket:
+            return
+
+        # Transcription frame will be only sent after we get the "endpoint" event.
+        self._final_transcription_buffer = []
+
+        async def send_endpoint_transcript():
+            if self._final_transcription_buffer:
+                text = "".join(map(lambda token: token["text"], self._final_transcription_buffer))
+                await self.push_frame(
+                    TranscriptionFrame(
+                        text=text,
+                        user_id=self._user_id,
+                        timestamp=time_now_iso8601(),
+                        result=self._final_transcription_buffer,
+                    )
+                )
+                await self._handle_transcription(text, is_final=True)
+                await self.stop_processing_metrics()
+                self._final_transcription_buffer = []
+
+        try:
+            async for message in self._websocket:
+                content = json.loads(message)
+
+                tokens = content["tokens"]
+
+                if tokens:
+                    if len(tokens) == 1 and tokens[0]["text"] == FINALIZED_TOKEN:
+                        # Ignore finalized token, prevent auto-finalize cycling.
+                        pass
+                    else:
+                        # Got at least one token, so we can reset the auto finalize delay.
+                        self._last_tokens_received = time.time()
+
+                # We will only send the final tokens after we get the "endpoint" event.
+                non_final_transcription = []
+
+                for token in tokens:
+                    if token["is_final"]:
+                        if is_end_token(token):
+                            # Found an endpoint, tokens until here will be sent as transcript,
+                            # the rest will be sent as interim tokens (even final tokens).
+                            await send_endpoint_transcript()
+                        else:
+                            self._final_transcription_buffer.append(token)
+                    else:
+                        non_final_transcription.append(token)
+
+                if self._final_transcription_buffer or non_final_transcription:
+                    final_text = "".join(
+                        map(lambda token: token["text"], self._final_transcription_buffer)
+                    )
+                    non_final_text = "".join(
+                        map(lambda token: token["text"], non_final_transcription)
+                    )
+
+                    await self.push_frame(
+                        InterimTranscriptionFrame(
+                            # Even final tokens are sent as interim tokens as we want to send
+                            # nicely formatted messages - therefore waiting for the endpoint.
+                            text=final_text + non_final_text,
+                            user_id=self._user_id,
+                            timestamp=time_now_iso8601(),
+                            result=self._final_transcription_buffer + non_final_transcription,
+                        )
+                    )
+
+                error_code = content.get("error_code")
+                error_message = content.get("error_message")
+                if error_code or error_message:
+                    # In case of error, still send the final transcript (if any remaining in the buffer).
+                    await send_endpoint_transcript()
+                    logger.error(
+                        f"{self} error: {error_code} (_receive_task_handler) - {error_message}"
+                    )
+                    await self.push_error(
+                        ErrorFrame(
+                            f"{self} error: {error_code} (_receive_task_handler) - {error_message}"
+                        )
+                    )
+
+                finished = content.get("finished")
+                if finished:
+                    # When finished, still send the final transcript (if any remaining in the buffer).
+                    await send_endpoint_transcript()
+                    logger.debug("Transcription finished.")
+                    await self._cleanup()
+                    return
+
+        except websockets.exceptions.ConnectionClosed:
+            # Expected when closing the connection.
+            pass
+        except Exception as e:
+            logger.error(f"{self} error: {e}")
+            await self.push_error(ErrorFrame(f"{self} error: {e}"))

src/pipecat/services/stt_service.py

@@ -152,6 +152,13 @@ class STTService(AIService):
         else:
             self._user_id = ""
 
+        if not frame.audio:
+            # Ignoring in case we don't have audio to transcribe.
+            logger.warning(
+                f"Empty audio frame received for STT service: {self.name} {frame.num_frames}"
+            )
+            return
+
         await self.process_generator(self.run_stt(frame.audio))
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):

src/pipecat/services/whisper/stt.py

@@ -49,8 +49,10 @@ class Model(Enum):
     Parameters:
         TINY: Smallest multilingual model, fastest inference.
         BASE: Basic multilingual model, good speed/quality balance.
+        SMALL: Small multilingual model, better speed/quality balance than BASE.
         MEDIUM: Medium-sized multilingual model, better quality.
         LARGE: Best quality multilingual model, slower inference.
+        LARGE_V3_TURBO: Fast multilingual model, slightly lower quality than LARGE.
         DISTIL_LARGE_V2: Fast multilingual distilled model.
         DISTIL_MEDIUM_EN: Fast English-only distilled model.
     """
@@ -58,8 +60,10 @@ class Model(Enum):
     # Multilingual models
     TINY = "tiny"
     BASE = "base"
+    SMALL = "small"
     MEDIUM = "medium"
     LARGE = "large-v3"
+    LARGE_V3_TURBO = "deepdml/faster-whisper-large-v3-turbo-ct2"
     DISTIL_LARGE_V2 = "Systran/faster-distil-whisper-large-v2"
 
     # English-only models

src/pipecat/transports/base_input.py

@@ -34,6 +34,7 @@ from pipecat.frames.frames import (
     InputAudioRawFrame,
     InputImageRawFrame,
     MetricsFrame,
+    SpeechControlParamsFrame,
     StartFrame,
     StartInterruptionFrame,
     StopFrame,
@@ -195,6 +196,13 @@ class BaseInputTransport(FrameProcessor):
         if self._params.turn_analyzer:
             self._params.turn_analyzer.set_sample_rate(self._sample_rate)
 
+        if self._params.vad_analyzer or self._params.turn_analyzer:
+            vad_params = self._params.vad_analyzer.params if self._params.vad_analyzer else None
+            turn_params = self._params.turn_analyzer.params if self._params.turn_analyzer else None
+
+            speech_frame = SpeechControlParamsFrame(vad_params=vad_params, turn_params=turn_params)
+            await self.push_frame(speech_frame)
+
         # Start audio filter.
         if self._params.audio_in_filter:
             await self._params.audio_in_filter.start(self._sample_rate)
@@ -310,6 +318,13 @@ class BaseInputTransport(FrameProcessor):
         elif isinstance(frame, VADParamsUpdateFrame):
             if self.vad_analyzer:
                 self.vad_analyzer.set_params(frame.params)
+                speech_frame = SpeechControlParamsFrame(
+                    vad_params=frame.params,
+                    turn_params=self._params.turn_analyzer.params
+                    if self._params.turn_analyzer
+                    else None,
+                )
+                await self.push_frame(speech_frame)
         elif isinstance(frame, FilterUpdateSettingsFrame) and self._params.audio_in_filter:
             await self._params.audio_in_filter.process_frame(frame)
         # Other frames

src/pipecat/transports/services/tavus.py

@@ -127,35 +127,6 @@ class TavusApi:
             logger.debug(f"Fetched Tavus persona: {response}")
             return response["persona_name"]
 
-    async def _validate_persona(self, persona_id: str):
-        """Validate that the persona's microphone is enabled.
-
-        Args:
-            persona_id: ID of the persona to validate.
-        """
-        if self._dev_room_url is not None:
-            return
-
-        url = f"{self.BASE_URL}/personas/{persona_id}"
-        async with self._session.get(url, headers=self._headers) as r:
-            r.raise_for_status()
-            response = await r.json()
-            logger.debug(f"Fetched Tavus persona: {response}")
-            try:
-                transport_settings = response.get("layers", {}).get("transport", {})
-                microphone_enabled = transport_settings.get("input_settings", {}).get(
-                    "microphone", ""
-                )
-                if microphone_enabled != "enabled":
-                    raise Exception(
-                        "Microphone is not enabled for this persona. Please update the persona or use the persona pipecat-stream."
-                    )
-            except Exception as e:
-                logger.error(f"Error validating persona {persona_id}: {e}")
-                raise e
-            logger.info(f"Persona {persona_id} is valid")
-            return True
-
 
 class TavusCallbacks(BaseModel):
     """Callback handlers for Tavus events.
@@ -229,7 +200,6 @@ class TavusTransportClient:
 
     async def _initialize(self) -> str:
         """Initialize the conversation and return the room URL."""
-        await self._api._validate_persona(self._persona_id)
         response = await self._api.create_conversation(self._replica_id, self._persona_id)
         self._conversation_id = response["conversation_id"]
         return response["conversation_url"]