Merge branch 'pipecat-ai:main' into anthropic-client-bug-fixes

src/pipecat/adapters/schemas/function_schema.py

@@ -8,16 +8,22 @@ from typing import Any, Dict, List
 
 
 class FunctionSchema:
+    """Standardized function schema representation for tool definition.
+
+    Provides a structured way to define function tools used with AI models like OpenAI.
+    This schema defines the function's name, description, parameter properties, and
+    required parameters, following specifications required by AI service providers.
+
+    Args:
+        name: Name of the function to be called.
+        description: Description of what the function does.
+        properties: Dictionary defining parameter types, descriptions, and constraints.
+        required: List of property names that are required parameters.
+    """
+
     def __init__(
         self, name: str, description: str, properties: Dict[str, Any], required: List[str]
     ) -> None:
-        """Standardized function schema representation.
-
-        :param name: Name of the function.
-        :param description: Description of the function.
-        :param properties: Dictionary defining properties types and descriptions.
-        :param required: List of required parameters.
-        """
         self._name = name
         self._description = description
         self._properties = properties
@@ -26,7 +32,8 @@ class FunctionSchema:
     def to_default_dict(self) -> Dict[str, Any]:
         """Converts the function schema to a dictionary.
 
-        :return: Dictionary representation of the function schema.
+        Returns:
+            Dictionary representation of the function schema.
         """
         return {
             "name": self._name,
@@ -40,16 +47,36 @@ class FunctionSchema:
 
     @property
     def name(self) -> str:
+        """Get the function name.
+
+        Returns:
+            The function name.
+        """
         return self._name
 
     @property
     def description(self) -> str:
+        """Get the function description.
+
+        Returns:
+            The function description.
+        """
         return self._description
 
     @property
     def properties(self) -> Dict[str, Any]:
+        """Get the function properties.
+
+        Returns:
+            Dictionary of parameter specifications.
+        """
         return self._properties
 
     @property
     def required(self) -> List[str]:
+        """Get the required parameters.
+
+        Returns:
+            List of required parameter names.
+        """
         return self._required

src/pipecat/frames/frames.py

@@ -377,25 +377,6 @@ class LLMEnablePromptCachingFrame(DataFrame):
     enable: bool
 
 
-@dataclass
-class FunctionCallResultProperties:
-    """Properties for a function call result frame."""
-
-    run_llm: Optional[bool] = None
-    on_context_updated: Optional[Callable[[], Awaitable[None]]] = None
-
-
-@dataclass
-class FunctionCallResultFrame(DataFrame):
-    """A frame containing the result of an LLM function (tool) call."""
-
-    function_name: str
-    tool_call_id: str
-    arguments: Any
-    result: Any
-    properties: Optional[FunctionCallResultProperties] = None
-
-
 @dataclass
 class TTSSpeakFrame(DataFrame):
     """A frame that contains a text that should be spoken by the TTS in the
@@ -652,6 +633,25 @@ class FunctionCallCancelFrame(SystemFrame):
     tool_call_id: str
 
 
+@dataclass
+class FunctionCallResultProperties:
+    """Properties for a function call result frame."""
+
+    run_llm: Optional[bool] = None
+    on_context_updated: Optional[Callable[[], Awaitable[None]]] = None
+
+
+@dataclass
+class FunctionCallResultFrame(SystemFrame):
+    """A frame containing the result of an LLM function (tool) call."""
+
+    function_name: str
+    tool_call_id: str
+    arguments: Any
+    result: Any
+    properties: Optional[FunctionCallResultProperties] = None
+
+
 @dataclass
 class STTMuteFrame(SystemFrame):
     """System frame to mute/unmute the STT service."""

src/pipecat/observers/loggers/llm_log_observer.py

@@ -18,7 +18,7 @@ from pipecat.frames.frames import (
 from pipecat.observers.base_observer import BaseObserver
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
-from pipecat.services.ai_services import LLMService
+from pipecat.services.llm_service import LLMService
 
 
 class LLMLogObserver(BaseObserver):

src/pipecat/observers/loggers/transcription_log_observer.py

@@ -13,7 +13,7 @@ from pipecat.frames.frames import (
 )
 from pipecat.observers.base_observer import BaseObserver
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
-from pipecat.services.ai_services import STTService
+from pipecat.services.stt_service import STTService
 
 
 class TranscriptionLogObserver(BaseObserver):

src/pipecat/processors/aggregators/llm_response.py

@@ -149,7 +149,8 @@ class BaseLLMResponseAggregator(FrameProcessor):
     @abstractmethod
     def reset(self):
         """Reset the internals of this aggregator. This should not modify the
-        internal messages."""
+        internal messages.
+        """
         pass
 
     @abstractmethod
@@ -446,6 +447,7 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
             await self._handle_user_image_frame(frame)
         elif isinstance(frame, BotStoppedSpeakingFrame):
             await self.push_aggregation()
+            await self.push_frame(frame, direction)
         else:
             await self.push_frame(frame, direction)
 

src/pipecat/processors/consumer_processor.py

@@ -0,0 +1,65 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+from typing import Awaitable, Callable, Optional
+
+from pipecat.frames.frames import CancelFrame, EndFrame, Frame, StartFrame
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.processors.producer_processor import ProducerProcessor, identity_transformer
+
+
+class ConsumerProcessor(FrameProcessor):
+    """This class passes-through frames and also consumes frames from a
+    producer's queue. When a frame from a producer queue is received it will be
+    pushed to the specified direction. The frames can be transformed into a
+    different type of frame before being pushed.
+
+    """
+
+    def __init__(
+        self,
+        *,
+        producer: ProducerProcessor,
+        transformer: Callable[[Frame], Awaitable[Frame]] = identity_transformer,
+        direction: FrameDirection = FrameDirection.DOWNSTREAM,
+        **kwargs,
+    ):
+        super().__init__(**kwargs)
+        self._transformer = transformer
+        self._direction = direction
+        self._queue: asyncio.Queue = producer.add_consumer()
+        self._consumer_task: Optional[asyncio.Task] = None
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, StartFrame):
+            await self._start(frame)
+        elif isinstance(frame, EndFrame):
+            await self._stop(frame)
+        elif isinstance(frame, CancelFrame):
+            await self._cancel(frame)
+
+        await self.push_frame(frame, direction)
+
+    async def _start(self, _: StartFrame):
+        if not self._consumer_task:
+            self._consumer_task = self.create_task(self._consumer_task_handler())
+
+    async def _stop(self, _: EndFrame):
+        if self._consumer_task:
+            await self.cancel_task(self._consumer_task)
+
+    async def _cancel(self, _: CancelFrame):
+        if self._consumer_task:
+            await self.cancel_task(self._consumer_task)
+
+    async def _consumer_task_handler(self):
+        while True:
+            frame = await self._queue.get()
+            new_frame = await self._transformer(frame)
+            await self.push_frame(new_frame, self._direction)

src/pipecat/processors/filters/stt_mute_filter.py

@@ -108,7 +108,7 @@ class STTMuteFilter(FrameProcessor):
     async def _handle_mute_state(self, should_mute: bool):
         """Handles both STT muting and interruption control."""
         if should_mute != self.is_muted:
-            logger.debug(f"STT {'muting' if should_mute else 'unmuting'}")
+            logger.debug(f"STTMuteFilter {'muting' if should_mute else 'unmuting'}")
             self._is_muted = should_mute
             await self.push_frame(STTMuteFrame(mute=should_mute))
 

src/pipecat/processors/producer_processor.py

@@ -0,0 +1,73 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+from typing import Awaitable, Callable, List
+
+from pipecat.frames.frames import Frame
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+
+
+async def identity_transformer(frame: Frame):
+    return frame
+
+
+class ProducerProcessor(FrameProcessor):
+    """This class optionally passes-through received frames and decides if those
+    frames should be sent to consumers based on a user-defined filter. The
+    frames can be transformed into a different type of frame before being
+    sending them to the consumers. More than one consumer can be added.
+
+    """
+
+    def __init__(
+        self,
+        *,
+        filter: Callable[[Frame], Awaitable[bool]],
+        transformer: Callable[[Frame], Awaitable[Frame]] = identity_transformer,
+        passthrough: bool = True,
+    ):
+        super().__init__()
+        self._filter = filter
+        self._transformer = transformer
+        self._passthrough = passthrough
+        self._consumers: List[asyncio.Queue] = []
+
+    def add_consumer(self):
+        """
+        Adds a new consumer and returns its associated queue.
+
+        Returns:
+            asyncio.Queue: The queue for the newly added consumer.
+        """
+        queue = asyncio.Queue()
+        self._consumers.append(queue)
+        return queue
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        """
+        Processes an incoming frame and determines whether to produce it as a ProducerItem.
+
+        If the frame meets the produce criteria, it will be added to the consumer queues.
+        If passthrough is enabled, the frame will also be sent to consumers.
+
+        Args:
+            frame (Frame): The frame to process.
+            direction (FrameDirection): The direction of the frame.
+        """
+        await super().process_frame(frame, direction)
+
+        if await self._filter(frame):
+            await self._produce(frame)
+            if self._passthrough:
+                await self.push_frame(frame, direction)
+        else:
+            await self.push_frame(frame, direction)
+
+    async def _produce(self, frame: Frame):
+        for consumer in self._consumers:
+            new_frame = await self._transformer(frame)
+            await consumer.put(new_frame)

src/pipecat/processors/transcript_processor.py

@@ -175,22 +175,28 @@ class AssistantTranscriptProcessor(BaseTranscriptProcessor):
         """
         await super().process_frame(frame, direction)
 
-        if isinstance(frame, TTSTextFrame):
+        if isinstance(frame, (StartInterruptionFrame, CancelFrame)):
+            # Push frame first otherwise our emitted transcription update frame
+            # might get cleaned up.
+            await self.push_frame(frame, direction)
+            # Emit accumulated text with interruptions
+            await self._emit_aggregated_text()
+        elif isinstance(frame, TTSTextFrame):
             # Start timestamp on first text part
             if not self._aggregation_start_time:
                 self._aggregation_start_time = time_now_iso8601()
 
             self._current_text_parts.append(frame.text)
 
-        elif isinstance(frame, (BotStoppedSpeakingFrame, StartInterruptionFrame, CancelFrame)):
-            # Emit accumulated text when bot finishes speaking or is interrupted
+            # Push frame.
+            await self.push_frame(frame, direction)
+        elif isinstance(frame, (BotStoppedSpeakingFrame, EndFrame)):
+            # Emit accumulated text when bot finishes speaking or pipeline ends.
             await self._emit_aggregated_text()
-
-        elif isinstance(frame, EndFrame):
-            # Emit any remaining text when pipeline ends
-            await self._emit_aggregated_text()
-
-        await self.push_frame(frame, direction)
+            # Push frame.
+            await self.push_frame(frame, direction)
+        else:
+            await self.push_frame(frame, direction)
 
 
 class TranscriptProcessor:

src/pipecat/processors/user_idle_processor.py

@@ -127,9 +127,10 @@ class UserIdleProcessor(FrameProcessor):
 
         # Check for end frames before processing
         if isinstance(frame, (EndFrame, CancelFrame)):
-            await self.push_frame(frame, direction)  # Push the frame down the pipeline
-            if self._idle_task:
-                await self._stop()  # Stop the idle task, if it exists
+            # Stop the idle task, if it exists
+            await self._stop()
+            # Push the frame down the pipeline
+            await self.push_frame(frame, direction)
             return
 
         await self.push_frame(frame, direction)

src/pipecat/services/__init__.py

@@ -37,4 +37,4 @@ class DeprecatedModuleProxy:
     def __getattr__(self, attr):
         if attr in self._globals:
             return _warn_deprecated_access(self._globals, attr, self._old, self._new)
-        raise AttributeError(f"module 'pipecat.{self._old}' has no attribute '{attr}'")
+        raise AttributeError(f"module 'pipecat.services.{self._old}' has no attribute '{attr}'")

src/pipecat/services/ai_service.py

@@ -0,0 +1,105 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+from typing import Any, AsyncGenerator, Dict, Mapping
+
+from loguru import logger
+
+from pipecat.frames.frames import (
+    CancelFrame,
+    EndFrame,
+    ErrorFrame,
+    Frame,
+    StartFrame,
+)
+from pipecat.metrics.metrics import MetricsData
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+
+
+class AIService(FrameProcessor):
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self._model_name: str = ""
+        self._settings: Dict[str, Any] = {}
+        self._session_properties: Dict[str, Any] = {}
+
+    @property
+    def model_name(self) -> str:
+        return self._model_name
+
+    def set_model_name(self, model: str):
+        self._model_name = model
+        self.set_core_metrics_data(MetricsData(processor=self.name, model=self._model_name))
+
+    async def start(self, frame: StartFrame):
+        pass
+
+    async def stop(self, frame: EndFrame):
+        pass
+
+    async def cancel(self, frame: CancelFrame):
+        pass
+
+    async def _update_settings(self, settings: Mapping[str, Any]):
+        from pipecat.services.openai_realtime_beta.events import (
+            SessionProperties,
+        )
+
+        for key, value in settings.items():
+            logger.debug("Update request for:", key, value)
+
+            if key in self._settings:
+                logger.info(f"Updating LLM setting {key} to: [{value}]")
+                self._settings[key] = value
+            elif key in SessionProperties.model_fields:
+                logger.debug("Attempting to update", key, value)
+
+                try:
+                    from pipecat.services.openai_realtime_beta.events import (
+                        TurnDetection,
+                    )
+
+                    if isinstance(self._session_properties, SessionProperties):
+                        current_properties = self._session_properties
+                    else:
+                        current_properties = SessionProperties(**self._session_properties)
+
+                    if key == "turn_detection" and isinstance(value, dict):
+                        turn_detection = TurnDetection(**value)
+                        setattr(current_properties, key, turn_detection)
+                    else:
+                        setattr(current_properties, key, value)
+
+                    validated_properties = SessionProperties.model_validate(
+                        current_properties.model_dump()
+                    )
+                    logger.info(f"Updating LLM setting {key} to: [{value}]")
+                    self._session_properties = validated_properties.model_dump()
+                except Exception as e:
+                    logger.warning(f"Unexpected error updating session property {key}: {e}")
+            elif key == "model":
+                logger.info(f"Updating LLM setting {key} to: [{value}]")
+                self.set_model_name(value)
+            else:
+                logger.warning(f"Unknown setting for {self.name} service: {key}")
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, StartFrame):
+            await self.start(frame)
+        elif isinstance(frame, CancelFrame):
+            await self.cancel(frame)
+        elif isinstance(frame, EndFrame):
+            await self.stop(frame)
+
+    async def process_generator(self, generator: AsyncGenerator[Frame | None, None]):
+        async for f in generator:
+            if f:
+                if isinstance(f, ErrorFrame):
+                    await self.push_error(f)
+                else:
+                    await self.push_frame(f)

src/pipecat/services/anthropic/llm.py

@@ -43,7 +43,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
     OpenAILLMContextFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
-from pipecat.services.ai_services import LLMService
+from pipecat.services.llm_service import LLMService
 
 try:
     from anthropic import NOT_GIVEN, AsyncAnthropic, NotGiven

src/pipecat/services/assemblyai/stt.py

@@ -18,7 +18,7 @@ from pipecat.frames.frames import (
     StartFrame,
     TranscriptionFrame,
 )
-from pipecat.services.ai_services import STTService
+from pipecat.services.stt_service import STTService
 from pipecat.transcriptions.language import Language
 from pipecat.utils.time import time_now_iso8601
 

src/pipecat/services/aws/tts.py

@@ -18,7 +18,7 @@ from pipecat.frames.frames import (
     TTSStartedFrame,
     TTSStoppedFrame,
 )
-from pipecat.services.ai_services import TTSService
+from pipecat.services.tts_service import TTSService
 from pipecat.transcriptions.language import Language
 
 try:

src/pipecat/services/azure/image.py

@@ -13,7 +13,7 @@ from loguru import logger
 from PIL import Image
 
 from pipecat.frames.frames import ErrorFrame, Frame, URLImageRawFrame
-from pipecat.services.ai_services import ImageGenService
+from pipecat.services.image_service import ImageGenService
 
 
 class AzureImageGenServiceREST(ImageGenService):

src/pipecat/services/azure/stt.py

@@ -16,8 +16,8 @@ from pipecat.frames.frames import (
     StartFrame,
     TranscriptionFrame,
 )
-from pipecat.services.ai_services import STTService
 from pipecat.services.azure.common import language_to_azure_language
+from pipecat.services.stt_service import STTService
 from pipecat.transcriptions.language import Language
 from pipecat.utils.time import time_now_iso8601
 

src/pipecat/services/azure/tts.py

@@ -18,8 +18,8 @@ from pipecat.frames.frames import (
     TTSStartedFrame,
     TTSStoppedFrame,
 )
-from pipecat.services.ai_services import TTSService
 from pipecat.services.azure.common import language_to_azure_language
+from pipecat.services.tts_service import TTSService
 from pipecat.transcriptions.language import Language
 
 try:

src/pipecat/services/canonical/metrics.py

@@ -18,7 +18,7 @@ from pipecat.frames.frames import CancelFrame, EndFrame, Frame
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor
 from pipecat.processors.frame_processor import FrameDirection
-from pipecat.services.ai_services import AIService
+from pipecat.services.ai_service import AIService
 
 try:
     import aiofiles

src/pipecat/services/cartesia/tts.py

@@ -24,7 +24,7 @@ from pipecat.frames.frames import (
     TTSStoppedFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
-from pipecat.services.ai_services import AudioContextWordTTSService, TTSService
+from pipecat.services.tts_service import AudioContextWordTTSService, TTSService
 from pipecat.transcriptions.language import Language
 from pipecat.utils.text.base_text_aggregator import BaseTextAggregator
 from pipecat.utils.text.skip_tags_aggregator import SkipTagsAggregator

src/pipecat/services/deepgram/stt.py

@@ -19,7 +19,7 @@ from pipecat.frames.frames import (
     UserStoppedSpeakingFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
-from pipecat.services.ai_services import STTService
+from pipecat.services.stt_service import STTService
 from pipecat.transcriptions.language import Language
 from pipecat.utils.time import time_now_iso8601
 

src/pipecat/services/deepgram/tts.py

@@ -16,7 +16,7 @@ from pipecat.frames.frames import (
     TTSStartedFrame,
     TTSStoppedFrame,
 )
-from pipecat.services.ai_services import TTSService
+from pipecat.services.tts_service import TTSService
 
 try:
     from deepgram import DeepgramClient, SpeakOptions

src/pipecat/services/elevenlabs/tts.py

@@ -25,7 +25,7 @@ from pipecat.frames.frames import (
     TTSStoppedFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
-from pipecat.services.ai_services import InterruptibleWordTTSService, TTSService
+from pipecat.services.tts_service import InterruptibleWordTTSService, TTSService
 from pipecat.transcriptions.language import Language
 
 # See .env.example for ElevenLabs configuration needed

src/pipecat/services/fal/image.py

@@ -15,7 +15,7 @@ from PIL import Image
 from pydantic import BaseModel
 
 from pipecat.frames.frames import ErrorFrame, Frame, URLImageRawFrame
-from pipecat.services.ai_services import ImageGenService
+from pipecat.services.image_service import ImageGenService
 
 try:
     import fal_client

src/pipecat/services/fal/stt.py

@@ -11,7 +11,7 @@ from loguru import logger
 from pydantic import BaseModel
 
 from pipecat.frames.frames import ErrorFrame, Frame, TranscriptionFrame
-from pipecat.services.ai_services import SegmentedSTTService
+from pipecat.services.stt_service import SegmentedSTTService
 from pipecat.transcriptions.language import Language
 from pipecat.utils.time import time_now_iso8601
 

src/pipecat/services/fish/tts.py

@@ -22,7 +22,7 @@ from pipecat.frames.frames import (
     TTSStoppedFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
-from pipecat.services.ai_services import InterruptibleTTSService
+from pipecat.services.tts_service import InterruptibleTTSService
 from pipecat.transcriptions.language import Language
 
 try:

src/pipecat/services/gemini_multimodal_live/gemini.py

@@ -50,7 +50,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
     OpenAILLMContextFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
-from pipecat.services.ai_services import LLMService
+from pipecat.services.llm_service import LLMService
 from pipecat.services.openai.llm import (
     OpenAIAssistantContextAggregator,
     OpenAIUserContextAggregator,

src/pipecat/services/gladia.py

@@ -1,253 +0,0 @@
-#
-# Copyright (c) 2024–2025, Daily
-#
-# SPDX-License-Identifier: BSD 2-Clause License
-#
-
-import base64
-import json
-from typing import AsyncGenerator, Optional
-
-import aiohttp
-from loguru import logger
-from pydantic import BaseModel
-
-from pipecat.frames.frames import (
-    CancelFrame,
-    EndFrame,
-    Frame,
-    InterimTranscriptionFrame,
-    StartFrame,
-    TranscriptionFrame,
-)
-from pipecat.services.ai_services import STTService
-from pipecat.transcriptions.language import Language
-from pipecat.utils.time import time_now_iso8601
-
-try:
-    import websockets
-except ModuleNotFoundError as e:
-    logger.error(f"Exception: {e}")
-    logger.error(
-        "In order to use Gladia, you need to `pip install pipecat-ai[gladia]`. Also, set `GLADIA_API_KEY` environment variable."
-    )
-    raise Exception(f"Missing module: {e}")
-
-
-def language_to_gladia_language(language: Language) -> Optional[str]:
-    BASE_LANGUAGES = {
-        Language.AF: "af",
-        Language.AM: "am",
-        Language.AR: "ar",
-        Language.AS: "as",
-        Language.AZ: "az",
-        Language.BG: "bg",
-        Language.BN: "bn",
-        Language.BS: "bs",
-        Language.CA: "ca",
-        Language.CS: "cs",
-        Language.CY: "cy",
-        Language.DA: "da",
-        Language.DE: "de",
-        Language.EL: "el",
-        Language.EN: "en",
-        Language.ES: "es",
-        Language.ET: "et",
-        Language.EU: "eu",
-        Language.FA: "fa",
-        Language.FI: "fi",
-        Language.FR: "fr",
-        Language.GA: "ga",
-        Language.GL: "gl",
-        Language.GU: "gu",
-        Language.HE: "he",
-        Language.HI: "hi",
-        Language.HR: "hr",
-        Language.HU: "hu",
-        Language.HY: "hy",
-        Language.ID: "id",
-        Language.IS: "is",
-        Language.IT: "it",
-        Language.JA: "ja",
-        Language.JV: "jv",
-        Language.KA: "ka",
-        Language.KK: "kk",
-        Language.KM: "km",
-        Language.KN: "kn",
-        Language.KO: "ko",
-        Language.LO: "lo",
-        Language.LT: "lt",
-        Language.LV: "lv",
-        Language.MK: "mk",
-        Language.ML: "ml",
-        Language.MN: "mn",
-        Language.MR: "mr",
-        Language.MS: "ms",
-        Language.MT: "mt",
-        Language.MY: "my",
-        Language.NE: "ne",
-        Language.NL: "nl",
-        Language.NO: "no",
-        Language.OR: "or",
-        Language.PA: "pa",
-        Language.PL: "pl",
-        Language.PS: "ps",
-        Language.PT: "pt",
-        Language.RO: "ro",
-        Language.RU: "ru",
-        Language.SI: "si",
-        Language.SK: "sk",
-        Language.SL: "sl",
-        Language.SO: "so",
-        Language.SQ: "sq",
-        Language.SR: "sr",
-        Language.SU: "su",
-        Language.SV: "sv",
-        Language.SW: "sw",
-        Language.TA: "ta",
-        Language.TE: "te",
-        Language.TH: "th",
-        Language.TR: "tr",
-        Language.UK: "uk",
-        Language.UR: "ur",
-        Language.UZ: "uz",
-        Language.VI: "vi",
-        Language.ZH: "zh",
-        Language.ZU: "zu",
-    }
-
-    result = BASE_LANGUAGES.get(language)
-
-    # If not found in base languages, try to find the base language from a variant
-    if not result:
-        # Convert enum value to string and get the base language part (e.g. es-ES -> es)
-        lang_str = str(language.value)
-        base_code = lang_str.split("-")[0].lower()
-        # Look up the base code in our supported languages
-        result = base_code if base_code in BASE_LANGUAGES.values() else None
-
-    return result
-
-
-class GladiaSTTService(STTService):
-    class InputParams(BaseModel):
-        language: Optional[Language] = Language.EN
-        endpointing: Optional[float] = 0.2
-        maximum_duration_without_endpointing: Optional[int] = 10
-        audio_enhancer: Optional[bool] = None
-        words_accurate_timestamps: Optional[bool] = None
-        speech_threshold: Optional[float] = 0.99
-
-    def __init__(
-        self,
-        *,
-        api_key: str,
-        url: str = "https://api.gladia.io/v2/live",
-        confidence: float = 0.5,
-        sample_rate: Optional[int] = None,
-        params: InputParams = InputParams(),
-        **kwargs,
-    ):
-        super().__init__(sample_rate=sample_rate, **kwargs)
-        self._api_key = api_key
-        self._url = url
-        self._settings = {
-            "encoding": "wav/pcm",
-            "bit_depth": 16,
-            "sample_rate": 0,
-            "channels": 1,
-            "language_config": {
-                "languages": [self.language_to_service_language(params.language)]
-                if params.language
-                else [],
-                "code_switching": False,
-            },
-            "endpointing": params.endpointing,
-            "maximum_duration_without_endpointing": params.maximum_duration_without_endpointing,
-            "pre_processing": {
-                "audio_enhancer": params.audio_enhancer,
-                "speech_threshold": params.speech_threshold,
-            },
-            "realtime_processing": {
-                "words_accurate_timestamps": params.words_accurate_timestamps,
-            },
-        }
-        self._confidence = confidence
-        self._websocket = None
-        self._receive_task = None
-
-    def language_to_service_language(self, language: Language) -> Optional[str]:
-        return language_to_gladia_language(language)
-
-    async def start(self, frame: StartFrame):
-        await super().start(frame)
-        if self._websocket:
-            return
-        self._settings["sample_rate"] = self.sample_rate
-        response = await self._setup_gladia()
-        self._websocket = await websockets.connect(response["url"])
-        if not self._receive_task:
-            self._receive_task = self.create_task(self._receive_task_handler())
-
-    async def stop(self, frame: EndFrame):
-        await super().stop(frame)
-        await self._send_stop_recording()
-        if self._websocket:
-            await self._websocket.close()
-            self._websocket = None
-        if self._receive_task:
-            await self.wait_for_task(self._receive_task)
-            self._receive_task = None
-
-    async def cancel(self, frame: CancelFrame):
-        await super().cancel(frame)
-        await self._websocket.close()
-        if self._receive_task:
-            await self.cancel_task(self._receive_task)
-            self._receive_task = None
-
-    async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
-        await self.start_processing_metrics()
-        await self._send_audio(audio)
-        await self.stop_processing_metrics()
-        yield None
-
-    async def _setup_gladia(self):
-        async with aiohttp.ClientSession() as session:
-            async with session.post(
-                self._url,
-                headers={"X-Gladia-Key": self._api_key, "Content-Type": "application/json"},
-                json=self._settings,
-            ) as response:
-                if response.ok:
-                    return await response.json()
-                else:
-                    logger.error(
-                        f"Gladia error: {response.status}: {response.text or response.reason}"
-                    )
-                    raise Exception(f"Failed to initialize Gladia session: {response.status}")
-
-    async def _send_audio(self, audio: bytes):
-        data = base64.b64encode(audio).decode("utf-8")
-        message = {"type": "audio_chunk", "data": {"chunk": data}}
-        await self._websocket.send(json.dumps(message))
-
-    async def _send_stop_recording(self):
-        await self._websocket.send(json.dumps({"type": "stop_recording"}))
-
-    async def _receive_task_handler(self):
-        async for message in self._websocket:
-            content = json.loads(message)
-            if content["type"] == "transcript":
-                utterance = content["data"]["utterance"]
-                confidence = utterance.get("confidence", 0)
-                transcript = utterance["text"]
-                if confidence >= self._confidence:
-                    if content["data"]["is_final"]:
-                        await self.push_frame(
-                            TranscriptionFrame(transcript, "", time_now_iso8601())
-                        )
-                    else:
-                        await self.push_frame(
-                            InterimTranscriptionFrame(transcript, "", time_now_iso8601())
-                        )

src/pipecat/services/gladia/__init__.py

@@ -0,0 +1,13 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import sys
+
+from pipecat.services import DeprecatedModuleProxy
+
+from .stt import *
+
+sys.modules[__name__] = DeprecatedModuleProxy(globals(), "gladia", "gladia.stt")

src/pipecat/services/gladia/config.py

@@ -0,0 +1,163 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+from typing import Any, Dict, List, Optional, Union
+
+from pydantic import BaseModel
+
+from pipecat.transcriptions.language import Language
+
+
+class LanguageConfig(BaseModel):
+    """Configuration for language detection and handling.
+
+    Attributes:
+        languages: List of language codes to use for transcription
+        code_switching: Whether to auto-detect language changes during transcription
+    """
+
+    languages: Optional[List[str]] = None
+    code_switching: Optional[bool] = None
+
+
+class PreProcessingConfig(BaseModel):
+    """Configuration for audio pre-processing options.
+
+    Attributes:
+        speech_threshold: Sensitivity for speech detection (0-1)
+    """
+
+    speech_threshold: Optional[float] = None
+
+
+class CustomVocabularyItem(BaseModel):
+    """Represents a custom vocabulary item with an intensity value.
+
+    Attributes:
+        value: The vocabulary word or phrase
+        intensity: The bias intensity for this vocabulary item (0-1)
+    """
+
+    value: str
+    intensity: float
+
+
+class CustomVocabularyConfig(BaseModel):
+    """Configuration for custom vocabulary.
+
+    Attributes:
+        vocabulary: List of words/phrases or CustomVocabularyItem objects
+        default_intensity: Default intensity for simple string vocabulary items
+    """
+
+    vocabulary: Optional[List[Union[str, CustomVocabularyItem]]] = None
+    default_intensity: Optional[float] = None
+
+
+class CustomSpellingConfig(BaseModel):
+    """Configuration for custom spelling rules.
+
+    Attributes:
+        spelling_dictionary: Mapping of correct spellings to phonetic variations
+    """
+
+    spelling_dictionary: Optional[Dict[str, List[str]]] = None
+
+
+class TranslationConfig(BaseModel):
+    """Configuration for real-time translation.
+
+    Attributes:
+        target_languages: List of target language codes for translation
+        model: Translation model to use ("base" or "enhanced")
+        match_original_utterances: Whether to align translations with original utterances
+    """
+
+    target_languages: Optional[List[str]] = None
+    model: Optional[str] = None
+    match_original_utterances: Optional[bool] = None
+
+
+class RealtimeProcessingConfig(BaseModel):
+    """Configuration for real-time processing features.
+
+    Attributes:
+        words_accurate_timestamps: Whether to provide per-word timestamps
+        custom_vocabulary: Whether to enable custom vocabulary
+        custom_vocabulary_config: Custom vocabulary configuration
+        custom_spelling: Whether to enable custom spelling
+        custom_spelling_config: Custom spelling configuration
+        translation: Whether to enable translation
+        translation_config: Translation configuration
+        named_entity_recognition: Whether to enable named entity recognition
+        sentiment_analysis: Whether to enable sentiment analysis
+    """
+
+    words_accurate_timestamps: Optional[bool] = None
+    custom_vocabulary: Optional[bool] = None
+    custom_vocabulary_config: Optional[CustomVocabularyConfig] = None
+    custom_spelling: Optional[bool] = None
+    custom_spelling_config: Optional[CustomSpellingConfig] = None
+    translation: Optional[bool] = None
+    translation_config: Optional[TranslationConfig] = None
+    named_entity_recognition: Optional[bool] = None
+    sentiment_analysis: Optional[bool] = None
+
+
+class MessagesConfig(BaseModel):
+    """Configuration for controlling which message types are sent via WebSocket.
+
+    Attributes:
+        receive_partial_transcripts: Whether to receive intermediate transcription results
+        receive_final_transcripts: Whether to receive final transcription results
+        receive_speech_events: Whether to receive speech begin/end events
+        receive_pre_processing_events: Whether to receive pre-processing events
+        receive_realtime_processing_events: Whether to receive real-time processing events
+        receive_post_processing_events: Whether to receive post-processing events
+        receive_acknowledgments: Whether to receive acknowledgment messages
+        receive_errors: Whether to receive error messages
+        receive_lifecycle_events: Whether to receive lifecycle events
+    """
+
+    receive_partial_transcripts: Optional[bool] = None
+    receive_final_transcripts: Optional[bool] = None
+    receive_speech_events: Optional[bool] = None
+    receive_pre_processing_events: Optional[bool] = None
+    receive_realtime_processing_events: Optional[bool] = None
+    receive_post_processing_events: Optional[bool] = None
+    receive_acknowledgments: Optional[bool] = None
+    receive_errors: Optional[bool] = None
+    receive_lifecycle_events: Optional[bool] = None
+
+
+class GladiaInputParams(BaseModel):
+    """Configuration parameters for the Gladia STT service.
+
+    Attributes:
+        encoding: Audio encoding format
+        bit_depth: Audio bit depth
+        channels: Number of audio channels
+        custom_metadata: Additional metadata to include with requests
+        endpointing: Silence duration in seconds to mark end of speech
+        maximum_duration_without_endpointing: Maximum utterance duration without silence
+        language: DEPRECATED - Use language_config instead
+        language_config: Detailed language configuration
+        pre_processing: Audio pre-processing options
+        realtime_processing: Real-time processing features
+        messages_config: WebSocket message filtering options
+    """
+
+    encoding: Optional[str] = "wav/pcm"
+    bit_depth: Optional[int] = 16
+    channels: Optional[int] = 1
+    custom_metadata: Optional[Dict[str, Any]] = None
+    endpointing: Optional[float] = None
+    maximum_duration_without_endpointing: Optional[int] = 10
+    language: Optional[Language] = None  # Deprecated
+    language_config: Optional[LanguageConfig] = None
+    pre_processing: Optional[PreProcessingConfig] = None
+    realtime_processing: Optional[RealtimeProcessingConfig] = None
+    messages_config: Optional[MessagesConfig] = None

src/pipecat/services/gladia/stt.py

@@ -0,0 +1,365 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import base64
+import json
+import warnings
+from typing import Any, AsyncGenerator, Dict, Optional
+
+import aiohttp
+from loguru import logger
+
+from pipecat.frames.frames import (
+    CancelFrame,
+    EndFrame,
+    Frame,
+    InterimTranscriptionFrame,
+    StartFrame,
+    TranscriptionFrame,
+)
+from pipecat.services.gladia.config import GladiaInputParams
+from pipecat.services.stt_service import STTService
+from pipecat.transcriptions.language import Language
+from pipecat.utils.time import time_now_iso8601
+
+try:
+    import websockets
+except ModuleNotFoundError as e:
+    logger.error(f"Exception: {e}")
+    logger.error("In order to use Gladia, you need to `pip install pipecat-ai[gladia]`.")
+    raise Exception(f"Missing module: {e}")
+
+
+def language_to_gladia_language(language: Language) -> Optional[str]:
+    """Convert a Language enum to Gladia's language code format.
+
+    Args:
+        language: The Language enum value to convert
+
+    Returns:
+        The Gladia language code string or None if not supported
+    """
+    BASE_LANGUAGES = {
+        Language.AF: "af",
+        Language.AM: "am",
+        Language.AR: "ar",
+        Language.AS: "as",
+        Language.AZ: "az",
+        Language.BA: "ba",
+        Language.BE: "be",
+        Language.BG: "bg",
+        Language.BN: "bn",
+        Language.BO: "bo",
+        Language.BR: "br",
+        Language.BS: "bs",
+        Language.CA: "ca",
+        Language.CS: "cs",
+        Language.CY: "cy",
+        Language.DA: "da",
+        Language.DE: "de",
+        Language.EL: "el",
+        Language.EN: "en",
+        Language.ES: "es",
+        Language.ET: "et",
+        Language.EU: "eu",
+        Language.FA: "fa",
+        Language.FI: "fi",
+        Language.FO: "fo",
+        Language.FR: "fr",
+        Language.GL: "gl",
+        Language.GU: "gu",
+        Language.HA: "ha",
+        Language.HAW: "haw",
+        Language.HE: "he",
+        Language.HI: "hi",
+        Language.HR: "hr",
+        Language.HT: "ht",
+        Language.HU: "hu",
+        Language.HY: "hy",
+        Language.ID: "id",
+        Language.IS: "is",
+        Language.IT: "it",
+        Language.JA: "ja",
+        Language.JV: "jv",
+        Language.KA: "ka",
+        Language.KK: "kk",
+        Language.KM: "km",
+        Language.KN: "kn",
+        Language.KO: "ko",
+        Language.LA: "la",
+        Language.LB: "lb",
+        Language.LN: "ln",
+        Language.LO: "lo",
+        Language.LT: "lt",
+        Language.LV: "lv",
+        Language.MG: "mg",
+        Language.MI: "mi",
+        Language.MK: "mk",
+        Language.ML: "ml",
+        Language.MN: "mn",
+        Language.MR: "mr",
+        Language.MS: "ms",
+        Language.MT: "mt",
+        Language.MY_MR: "mymr",
+        Language.NE: "ne",
+        Language.NL: "nl",
+        Language.NN: "nn",
+        Language.NO: "no",
+        Language.OC: "oc",
+        Language.PA: "pa",
+        Language.PL: "pl",
+        Language.PS: "ps",
+        Language.PT: "pt",
+        Language.RO: "ro",
+        Language.RU: "ru",
+        Language.SA: "sa",
+        Language.SD: "sd",
+        Language.SI: "si",
+        Language.SK: "sk",
+        Language.SL: "sl",
+        Language.SN: "sn",
+        Language.SO: "so",
+        Language.SQ: "sq",
+        Language.SR: "sr",
+        Language.SU: "su",
+        Language.SV: "sv",
+        Language.SW: "sw",
+        Language.TA: "ta",
+        Language.TE: "te",
+        Language.TG: "tg",
+        Language.TH: "th",
+        Language.TK: "tk",
+        Language.TL: "tl",
+        Language.TR: "tr",
+        Language.TT: "tt",
+        Language.UK: "uk",
+        Language.UR: "ur",
+        Language.UZ: "uz",
+        Language.VI: "vi",
+        Language.YI: "yi",
+        Language.YO: "yo",
+        Language.ZH: "zh",
+    }
+
+    result = BASE_LANGUAGES.get(language)
+
+    # If not found in base languages, try to find the base language from a variant
+    if not result:
+        # Convert enum value to string and get the base language part (e.g. es-ES -> es)
+        lang_str = str(language.value)
+        base_code = lang_str.split("-")[0].lower()
+        # Look up the base code in our supported languages
+        result = base_code if base_code in BASE_LANGUAGES.values() else None
+
+    return result
+
+
+# Deprecation warning for nested InputParams
+class _InputParamsDescriptor:
+    """Descriptor for backward compatibility with deprecation warning."""
+
+    def __get__(self, obj, objtype=None):
+        warnings.warn(
+            "GladiaSTTService.InputParams is deprecated and will be removed in a future version. "
+            "Import and use GladiaInputParams directly instead.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+        return GladiaInputParams
+
+
+class GladiaSTTService(STTService):
+    """Speech-to-Text service using Gladia's API.
+
+    This service connects to Gladia's WebSocket API for real-time transcription
+    with support for multiple languages, custom vocabulary, and various processing options.
+
+    For complete API documentation, see: https://docs.gladia.io/api-reference/v2/live/init
+    """
+
+    # Maintain backward compatibility
+    InputParams = _InputParamsDescriptor()
+
+    def __init__(
+        self,
+        *,
+        api_key: str,
+        url: str = "https://api.gladia.io/v2/live",
+        confidence: float = 0.5,
+        sample_rate: Optional[int] = None,
+        model: str = "solaria-1",
+        params: GladiaInputParams = GladiaInputParams(),
+        **kwargs,
+    ):
+        """Initialize the Gladia STT service.
+
+        Args:
+            api_key: Gladia API key
+            url: Gladia API URL
+            confidence: Minimum confidence threshold for transcriptions
+            sample_rate: Audio sample rate in Hz
+            model: Model to use ("solaria-1", "solaria-mini-1", "fast",
+                or "accurate")
+            params: Additional configuration parameters
+            **kwargs: Additional arguments passed to the STTService
+        """
+        super().__init__(sample_rate=sample_rate, **kwargs)
+
+        # Warn about deprecated language parameter if it's used
+        if params.language is not None:
+            warnings.warn(
+                "The 'language' parameter is deprecated and will be removed in a future version. "
+                "Use 'language_config' instead.",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+
+        self._api_key = api_key
+        self._url = url
+        self.set_model_name(model)
+        self._confidence = confidence
+        self._params = params
+        self._websocket = None
+        self._receive_task = None
+
+    def language_to_service_language(self, language: Language) -> Optional[str]:
+        """Convert pipecat Language enum to Gladia's language code."""
+        return language_to_gladia_language(language)
+
+    def _prepare_settings(self) -> Dict[str, Any]:
+        settings = {
+            "encoding": self._params.encoding or "wav/pcm",
+            "bit_depth": self._params.bit_depth or 16,
+            "sample_rate": self.sample_rate,
+            "channels": self._params.channels or 1,
+            "model": self._model_name,
+        }
+
+        # Add custom_metadata if provided
+        if self._params.custom_metadata:
+            settings["custom_metadata"] = self._params.custom_metadata
+
+        # Add endpointing parameters if provided
+        if self._params.endpointing is not None:
+            settings["endpointing"] = self._params.endpointing
+        if self._params.maximum_duration_without_endpointing is not None:
+            settings["maximum_duration_without_endpointing"] = (
+                self._params.maximum_duration_without_endpointing
+            )
+
+        # Add language configuration (prioritize language_config over deprecated language)
+        if self._params.language_config:
+            settings["language_config"] = self._params.language_config.model_dump(exclude_none=True)
+        elif self._params.language:  # Backward compatibility for deprecated parameter
+            language_code = self.language_to_service_language(self._params.language)
+            if language_code:
+                settings["language_config"] = {
+                    "languages": [language_code],
+                    "code_switching": False,
+                }
+
+        # Add pre_processing configuration if provided
+        if self._params.pre_processing:
+            settings["pre_processing"] = self._params.pre_processing.model_dump(exclude_none=True)
+
+        # Add realtime_processing configuration if provided
+        if self._params.realtime_processing:
+            settings["realtime_processing"] = self._params.realtime_processing.model_dump(
+                exclude_none=True
+            )
+
+        # Add messages_config if provided
+        if self._params.messages_config:
+            settings["messages_config"] = self._params.messages_config.model_dump(exclude_none=True)
+
+        return settings
+
+    async def start(self, frame: StartFrame):
+        """Start the Gladia STT websocket connection."""
+        await super().start(frame)
+        if self._websocket:
+            return
+        settings = self._prepare_settings()
+        response = await self._setup_gladia(settings)
+        self._websocket = await websockets.connect(response["url"])
+        if not self._receive_task:
+            self._receive_task = self.create_task(self._receive_task_handler())
+
+    async def stop(self, frame: EndFrame):
+        """Stop the Gladia STT websocket connection."""
+        await super().stop(frame)
+        await self._send_stop_recording()
+        if self._websocket:
+            await self._websocket.close()
+            self._websocket = None
+        if self._receive_task:
+            await self.wait_for_task(self._receive_task)
+            self._receive_task = None
+
+    async def cancel(self, frame: CancelFrame):
+        """Cancel the Gladia STT websocket connection."""
+        await super().cancel(frame)
+        await self._websocket.close()
+        if self._receive_task:
+            await self.cancel_task(self._receive_task)
+            self._receive_task = None
+
+    async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
+        """Run speech-to-text on audio data."""
+        await self.start_processing_metrics()
+        await self._send_audio(audio)
+        await self.stop_processing_metrics()
+        yield None
+
+    async def _setup_gladia(self, settings: Dict[str, Any]):
+        async with aiohttp.ClientSession() as session:
+            async with session.post(
+                self._url,
+                headers={"X-Gladia-Key": self._api_key, "Content-Type": "application/json"},
+                json=settings,
+            ) as response:
+                if response.ok:
+                    return await response.json()
+                else:
+                    error_text = await response.text()
+                    logger.error(
+                        f"Gladia error: {response.status}: {error_text or response.reason}"
+                    )
+                    raise Exception(
+                        f"Failed to initialize Gladia session: {response.status} - {error_text}"
+                    )
+
+    async def _send_audio(self, audio: bytes):
+        data = base64.b64encode(audio).decode("utf-8")
+        message = {"type": "audio_chunk", "data": {"chunk": data}}
+        await self._websocket.send(json.dumps(message))
+
+    async def _send_stop_recording(self):
+        if self._websocket and not self._websocket.closed:
+            await self._websocket.send(json.dumps({"type": "stop_recording"}))
+
+    async def _receive_task_handler(self):
+        try:
+            async for message in self._websocket:
+                content = json.loads(message)
+                if content["type"] == "transcript":
+                    utterance = content["data"]["utterance"]
+                    confidence = utterance.get("confidence", 0)
+                    transcript = utterance["text"]
+                    if confidence >= self._confidence:
+                        if content["data"]["is_final"]:
+                            await self.push_frame(
+                                TranscriptionFrame(transcript, "", time_now_iso8601())
+                            )
+                        else:
+                            await self.push_frame(
+                                InterimTranscriptionFrame(transcript, "", time_now_iso8601())
+                            )
+        except websockets.exceptions.ConnectionClosed:
+            # Expected when closing the connection
+            pass
+        except Exception as e:
+            logger.error(f"Error in Gladia WebSocket handler: {e}")

src/pipecat/services/google/image.py

@@ -17,7 +17,7 @@ from PIL import Image
 from pydantic import BaseModel, Field
 
 from pipecat.frames.frames import ErrorFrame, Frame, URLImageRawFrame
-from pipecat.services.ai_services import ImageGenService
+from pipecat.services.image_service import ImageGenService
 
 try:
     from google import genai

src/pipecat/services/google/llm.py

@@ -44,8 +44,8 @@ from pipecat.processors.aggregators.openai_llm_context import (
     OpenAILLMContextFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
-from pipecat.services.ai_services import LLMService
 from pipecat.services.google.frames import LLMSearchResponseFrame
+from pipecat.services.llm_service import LLMService
 from pipecat.services.openai.llm import (
     OpenAIAssistantContextAggregator,
     OpenAIUserContextAggregator,

src/pipecat/services/google/stt.py

@@ -26,7 +26,7 @@ from pipecat.frames.frames import (
     StartFrame,
     TranscriptionFrame,
 )
-from pipecat.services.ai_services import STTService
+from pipecat.services.stt_service import STTService
 from pipecat.transcriptions.language import Language
 from pipecat.utils.time import time_now_iso8601
 

src/pipecat/services/google/tts.py

@@ -23,7 +23,7 @@ from pipecat.frames.frames import (
     TTSStartedFrame,
     TTSStoppedFrame,
 )
-from pipecat.services.ai_services import TTSService
+from pipecat.services.tts_service import TTSService
 from pipecat.transcriptions.language import Language
 
 try:

src/pipecat/services/groq/tts.py

@@ -10,7 +10,7 @@ from loguru import logger
 from pydantic import BaseModel
 
 from pipecat.frames.frames import Frame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame
-from pipecat.services.ai_services import TTSService
+from pipecat.services.tts_service import TTSService
 from pipecat.transcriptions.language import Language
 
 try:

src/pipecat/services/image_service.py

@@ -0,0 +1,33 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+from abc import abstractmethod
+from typing import AsyncGenerator
+
+from pipecat.frames.frames import Frame, TextFrame
+from pipecat.processors.frame_processor import FrameDirection
+from pipecat.services.ai_service import AIService
+
+
+class ImageGenService(AIService):
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+    # Renders the image. Returns an Image object.
+    @abstractmethod
+    async def run_image_gen(self, prompt: str) -> AsyncGenerator[Frame, None]:
+        pass
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, TextFrame):
+            await self.push_frame(frame, direction)
+            await self.start_processing_metrics()
+            await self.process_generator(self.run_image_gen(frame.text))
+            await self.stop_processing_metrics()
+        else:
+            await self.push_frame(frame, direction)

src/pipecat/services/llm_service.py

@@ -0,0 +1,257 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+from dataclasses import dataclass
+from typing import Any, Mapping, Optional, Set, Tuple, Type
+
+from loguru import logger
+
+from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
+from pipecat.adapters.services.open_ai_adapter import OpenAILLMAdapter
+from pipecat.frames.frames import (
+    Frame,
+    FunctionCallCancelFrame,
+    FunctionCallInProgressFrame,
+    FunctionCallResultFrame,
+    StartInterruptionFrame,
+    UserImageRequestFrame,
+)
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.frame_processor import FrameDirection
+from pipecat.services.ai_service import AIService
+
+
+@dataclass
+class FunctionEntry:
+    function_name: Optional[str]
+    callback: Any  # TODO(aleix): add proper typing.
+    cancel_on_interruption: bool
+
+
+class LLMService(AIService):
+    """This class is a no-op but serves as a base class for LLM services."""
+
+    # OpenAILLMAdapter is used as the default adapter since it aligns with most LLM implementations.
+    # However, subclasses should override this with a more specific adapter when necessary.
+    adapter_class: Type[BaseLLMAdapter] = OpenAILLMAdapter
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self._functions = {}
+        self._start_callbacks = {}
+        self._adapter = self.adapter_class()
+        self._function_call_tasks: Set[Tuple[asyncio.Task, str, str]] = set()
+
+        self._register_event_handler("on_completion_timeout")
+
+    def get_llm_adapter(self) -> BaseLLMAdapter:
+        return self._adapter
+
+    def create_context_aggregator(
+        self,
+        context: OpenAILLMContext,
+        *,
+        user_kwargs: Mapping[str, Any] = {},
+        assistant_kwargs: Mapping[str, Any] = {},
+    ) -> Any:
+        pass
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, StartInterruptionFrame):
+            await self._handle_interruptions(frame)
+
+    async def _handle_interruptions(self, frame: StartInterruptionFrame):
+        for function_name, entry in self._functions.items():
+            if entry.cancel_on_interruption:
+                await self._cancel_function_call(function_name)
+
+    def register_function(
+        self,
+        function_name: Optional[str],
+        callback: Any,
+        start_callback=None,
+        *,
+        cancel_on_interruption: bool = False,
+    ):
+        # Registering a function with the function_name set to None will run that callback
+        # for all functions
+        self._functions[function_name] = FunctionEntry(
+            function_name=function_name,
+            callback=callback,
+            cancel_on_interruption=cancel_on_interruption,
+        )
+
+        # Start callbacks are now deprecated.
+        if start_callback:
+            import warnings
+
+            with warnings.catch_warnings():
+                warnings.simplefilter("always")
+                warnings.warn(
+                    "Parameter 'start_callback' is deprecated, just put your code on top of the actual function call instead.",
+                    DeprecationWarning,
+                )
+
+            self._start_callbacks[function_name] = start_callback
+
+    def unregister_function(self, function_name: Optional[str]):
+        del self._functions[function_name]
+        if self._start_callbacks[function_name]:
+            del self._start_callbacks[function_name]
+
+    def has_function(self, function_name: str):
+        if None in self._functions.keys():
+            return True
+        return function_name in self._functions.keys()
+
+    async def call_function(
+        self,
+        *,
+        context: OpenAILLMContext,
+        tool_call_id: str,
+        function_name: str,
+        arguments: str,
+        run_llm: bool = True,
+    ):
+        if not function_name in self._functions.keys() and not None in self._functions.keys():
+            return
+
+        task = self.create_task(
+            self._run_function_call(context, tool_call_id, function_name, arguments, run_llm)
+        )
+
+        self._function_call_tasks.add((task, tool_call_id, function_name))
+
+        task.add_done_callback(self._function_call_task_finished)
+
+    async def call_start_function(self, context: OpenAILLMContext, function_name: str):
+        if function_name in self._start_callbacks.keys():
+            await self._start_callbacks[function_name](function_name, self, context)
+        elif None in self._start_callbacks.keys():
+            return await self._start_callbacks[None](function_name, self, context)
+
+    async def request_image_frame(
+        self,
+        user_id: str,
+        *,
+        function_name: Optional[str] = None,
+        tool_call_id: Optional[str] = None,
+        text_content: Optional[str] = None,
+    ):
+        await self.push_frame(
+            UserImageRequestFrame(
+                user_id=user_id,
+                function_name=function_name,
+                tool_call_id=tool_call_id,
+                context=text_content,
+            ),
+            FrameDirection.UPSTREAM,
+        )
+
+    async def _run_function_call(
+        self,
+        context: OpenAILLMContext,
+        tool_call_id: str,
+        function_name: str,
+        arguments: str,
+        run_llm: bool = True,
+    ):
+        if function_name in self._functions.keys():
+            entry = self._functions[function_name]
+        elif None in self._functions.keys():
+            entry = self._functions[None]
+        else:
+            return
+
+        logger.debug(
+            f"{self} Calling function [{function_name}:{tool_call_id}] with arguments {arguments}"
+        )
+
+        # NOTE(aleix): This needs to be removed after we remove the deprecation.
+        await self.call_start_function(context, function_name)
+
+        # Push a SystemFrame downstream. This frame will let our assistant context aggregator
+        # know that we are in the middle of a function call. Some contexts/aggregators may
+        # not need this. But some definitely do (Anthropic, for example).
+        # Also push a SystemFrame upstream for use by other processors, like STTMuteFilter.
+        progress_frame_downstream = FunctionCallInProgressFrame(
+            function_name=function_name,
+            tool_call_id=tool_call_id,
+            arguments=arguments,
+            cancel_on_interruption=entry.cancel_on_interruption,
+        )
+        progress_frame_upstream = FunctionCallInProgressFrame(
+            function_name=function_name,
+            tool_call_id=tool_call_id,
+            arguments=arguments,
+            cancel_on_interruption=entry.cancel_on_interruption,
+        )
+
+        # Push frame both downstream and upstream
+        await self.push_frame(progress_frame_downstream, FrameDirection.DOWNSTREAM)
+        await self.push_frame(progress_frame_upstream, FrameDirection.UPSTREAM)
+
+        # Define a callback function that pushes a FunctionCallResultFrame upstream & downstream.
+        async def function_call_result_callback(result, *, properties=None):
+            result_frame_downstream = FunctionCallResultFrame(
+                function_name=function_name,
+                tool_call_id=tool_call_id,
+                arguments=arguments,
+                result=result,
+                properties=properties,
+            )
+            result_frame_upstream = FunctionCallResultFrame(
+                function_name=function_name,
+                tool_call_id=tool_call_id,
+                arguments=arguments,
+                result=result,
+                properties=properties,
+            )
+
+            await self.push_frame(result_frame_downstream, FrameDirection.DOWNSTREAM)
+            await self.push_frame(result_frame_upstream, FrameDirection.UPSTREAM)
+
+        await entry.callback(
+            function_name, tool_call_id, arguments, self, context, function_call_result_callback
+        )
+
+    async def _cancel_function_call(self, function_name: str):
+        cancelled_tasks = set()
+        for task, tool_call_id, name in self._function_call_tasks:
+            if name == function_name:
+                # We remove the callback because we are going to cancel the task
+                # now, otherwise we will be removing it from the set while we
+                # are iterating.
+                task.remove_done_callback(self._function_call_task_finished)
+
+                logger.debug(f"{self} Cancelling function call [{name}:{tool_call_id}]...")
+
+                await self.cancel_task(task)
+
+                frame = FunctionCallCancelFrame(
+                    function_name=function_name, tool_call_id=tool_call_id
+                )
+                await self.push_frame(frame)
+
+                logger.debug(f"{self} Function call [{name}:{tool_call_id}] has been cancelled")
+
+                cancelled_tasks.add(task)
+
+        # Remove all cancelled tasks from our set.
+        for task in cancelled_tasks:
+            self._function_call_task_finished(task)
+
+    def _function_call_task_finished(self, task: asyncio.Task):
+        tuple_to_remove = next((t for t in self._function_call_tasks if t[0] == task), None)
+        if tuple_to_remove:
+            self._function_call_tasks.discard(tuple_to_remove)
+            # The task is finished so this should exit immediately. We need to
+            # do this because otherwise the task manager would report a dangling
+            # task if we don't remove it.
+            asyncio.run_coroutine_threadsafe(self.wait_for_task(task), self.get_event_loop())

src/pipecat/services/lmnt/tts.py

@@ -21,7 +21,7 @@ from pipecat.frames.frames import (
     TTSStoppedFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
-from pipecat.services.ai_services import InterruptibleTTSService
+from pipecat.services.tts_service import InterruptibleTTSService
 from pipecat.transcriptions.language import Language
 
 # See .env.example for LMNT configuration needed

src/pipecat/services/moondream/vision.py

@@ -11,7 +11,7 @@ from loguru import logger
 from PIL import Image
 
 from pipecat.frames.frames import ErrorFrame, Frame, TextFrame, VisionImageRawFrame
-from pipecat.services.ai_services import VisionService
+from pipecat.services.vision_service import VisionService
 
 try:
     import torch

src/pipecat/services/neuphonic/tts.py

@@ -27,7 +27,7 @@ from pipecat.frames.frames import (
     TTSStoppedFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
-from pipecat.services.ai_services import InterruptibleTTSService, TTSService
+from pipecat.services.tts_service import InterruptibleTTSService, TTSService
 from pipecat.transcriptions.language import Language
 
 try:

src/pipecat/services/openai/base_llm.py

@@ -34,7 +34,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
     OpenAILLMContextFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
-from pipecat.services.ai_services import LLMService
+from pipecat.services.llm_service import LLMService
 
 
 class OpenAIUnhandledFunctionException(Exception):

src/pipecat/services/openai/image.py

@@ -17,7 +17,7 @@ from pipecat.frames.frames import (
     Frame,
     URLImageRawFrame,
 )
-from pipecat.services.ai_services import ImageGenService
+from pipecat.services.image_service import ImageGenService
 
 
 class OpenAIImageGenService(ImageGenService):

src/pipecat/services/openai/tts.py

@@ -17,17 +17,24 @@ from pipecat.frames.frames import (
     TTSStartedFrame,
     TTSStoppedFrame,
 )
-from pipecat.services.ai_services import TTSService
+from pipecat.services.tts_service import TTSService
 
-ValidVoice = Literal["alloy", "echo", "fable", "onyx", "nova", "shimmer"]
+ValidVoice = Literal[
+    "alloy", "ash", "ballad", "coral", "echo", "fable", "onyx", "nova", "sage", "shimmer", "verse"
+]
 
 VALID_VOICES: Dict[str, ValidVoice] = {
     "alloy": "alloy",
+    "ash": "ash",
+    "ballad": "ballad",
+    "coral": "coral",
     "echo": "echo",
     "fable": "fable",
     "onyx": "onyx",
     "nova": "nova",
+    "sage": "sage",
     "shimmer": "shimmer",
+    "verse": "verse",
 }
 
 

src/pipecat/services/openai_realtime_beta/openai.py

@@ -53,7 +53,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
     OpenAILLMContextFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
-from pipecat.services.ai_services import LLMService
+from pipecat.services.llm_service import LLMService
 from pipecat.services.openai.llm import OpenAIContextAggregatorPair
 from pipecat.utils.time import time_now_iso8601
 

src/pipecat/services/piper/tts.py

@@ -16,7 +16,7 @@ from pipecat.frames.frames import (
     TTSStartedFrame,
     TTSStoppedFrame,
 )
-from pipecat.services.ai_services import TTSService
+from pipecat.services.tts_service import TTSService
 
 
 # This assumes a running TTS service running: https://github.com/rhasspy/piper/blob/master/src/python_run/README_http.md

src/pipecat/services/playht/tts.py

@@ -27,7 +27,7 @@ from pipecat.frames.frames import (
     TTSStoppedFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
-from pipecat.services.ai_services import InterruptibleTTSService, TTSService
+from pipecat.services.tts_service import InterruptibleTTSService, TTSService
 from pipecat.transcriptions.language import Language
 
 try:

src/pipecat/services/rime/tts.py

@@ -25,7 +25,7 @@ from pipecat.frames.frames import (
     TTSStoppedFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
-from pipecat.services.ai_services import AudioContextWordTTSService, TTSService
+from pipecat.services.tts_service import AudioContextWordTTSService, TTSService
 from pipecat.transcriptions.language import Language
 from pipecat.utils.text.base_text_aggregator import BaseTextAggregator
 from pipecat.utils.text.skip_tags_aggregator import SkipTagsAggregator

src/pipecat/services/riva/stt.py

@@ -18,7 +18,7 @@ from pipecat.frames.frames import (
     StartFrame,
     TranscriptionFrame,
 )
-from pipecat.services.ai_services import STTService
+from pipecat.services.stt_service import STTService
 from pipecat.transcriptions.language import Language
 from pipecat.utils.time import time_now_iso8601
 

src/pipecat/services/riva/tts.py

@@ -16,7 +16,7 @@ from pipecat.frames.frames import (
     TTSStartedFrame,
     TTSStoppedFrame,
 )
-from pipecat.services.ai_services import TTSService
+from pipecat.services.tts_service import TTSService
 from pipecat.transcriptions.language import Language
 
 try:

src/pipecat/services/stt_service.py

@@ -0,0 +1,171 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import io
+import wave
+from abc import abstractmethod
+from typing import Any, AsyncGenerator, Dict, Mapping, Optional
+
+from loguru import logger
+
+from pipecat.frames.frames import (
+    AudioRawFrame,
+    Frame,
+    StartFrame,
+    STTMuteFrame,
+    STTUpdateSettingsFrame,
+    UserStartedSpeakingFrame,
+    UserStoppedSpeakingFrame,
+)
+from pipecat.processors.frame_processor import FrameDirection
+from pipecat.services.ai_service import AIService
+from pipecat.transcriptions.language import Language
+
+
+class STTService(AIService):
+    """STTService is a base class for speech-to-text services."""
+
+    def __init__(
+        self,
+        audio_passthrough=False,
+        # STT input sample rate
+        sample_rate: Optional[int] = None,
+        **kwargs,
+    ):
+        super().__init__(**kwargs)
+        self._audio_passthrough = audio_passthrough
+        self._init_sample_rate = sample_rate
+        self._sample_rate = 0
+        self._settings: Dict[str, Any] = {}
+        self._muted: bool = False
+
+    @property
+    def is_muted(self) -> bool:
+        """Returns whether the STT service is currently muted."""
+        return self._muted
+
+    @property
+    def sample_rate(self) -> int:
+        return self._sample_rate
+
+    async def set_model(self, model: str):
+        self.set_model_name(model)
+
+    async def set_language(self, language: Language):
+        pass
+
+    @abstractmethod
+    async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
+        """Returns transcript as a string"""
+        pass
+
+    async def start(self, frame: StartFrame):
+        await super().start(frame)
+        self._sample_rate = self._init_sample_rate or frame.audio_in_sample_rate
+
+    async def _update_settings(self, settings: Mapping[str, Any]):
+        logger.info(f"Updating STT settings: {self._settings}")
+        for key, value in settings.items():
+            if key in self._settings:
+                logger.info(f"Updating STT setting {key} to: [{value}]")
+                self._settings[key] = value
+                if key == "language":
+                    await self.set_language(value)
+            elif key == "model":
+                self.set_model_name(value)
+            else:
+                logger.warning(f"Unknown setting for STT service: {key}")
+
+    async def process_audio_frame(self, frame: AudioRawFrame, direction: FrameDirection):
+        if self._muted:
+            return
+
+        await self.process_generator(self.run_stt(frame.audio))
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        """Processes a frame of audio data, either buffering or transcribing it."""
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, AudioRawFrame):
+            # In this service we accumulate audio internally and at the end we
+            # push a TextFrame. We also push audio downstream in case someone
+            # else needs it.
+            await self.process_audio_frame(frame, direction)
+            if self._audio_passthrough:
+                await self.push_frame(frame, direction)
+        elif isinstance(frame, STTUpdateSettingsFrame):
+            await self._update_settings(frame.settings)
+        elif isinstance(frame, STTMuteFrame):
+            self._muted = frame.mute
+            logger.debug(f"STT service {'muted' if frame.mute else 'unmuted'}")
+        else:
+            await self.push_frame(frame, direction)
+
+
+class SegmentedSTTService(STTService):
+    """SegmentedSTTService is an STTService that uses VAD events to detect
+    speech and will run speech-to-text on speech segments only, instead of a
+    continous stream. Since it uses VAD it means that VAD needs to be enabled in
+    the pipeline.
+
+    This service always keeps a small audio buffer to take into account that VAD
+    events are delayed from when the user speech really starts.
+
+    """
+
+    def __init__(self, *, sample_rate: Optional[int] = None, **kwargs):
+        super().__init__(sample_rate=sample_rate, **kwargs)
+        self._content = None
+        self._wave = None
+        self._audio_buffer = bytearray()
+        self._audio_buffer_size_1s = 0
+        self._user_speaking = False
+
+    async def start(self, frame: StartFrame):
+        await super().start(frame)
+        self._audio_buffer_size_1s = self.sample_rate * 2
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, UserStartedSpeakingFrame):
+            await self._handle_user_started_speaking(frame)
+        elif isinstance(frame, UserStoppedSpeakingFrame):
+            await self._handle_user_stopped_speaking(frame)
+
+    async def _handle_user_started_speaking(self, frame: UserStartedSpeakingFrame):
+        if frame.emulated:
+            return
+        self._user_speaking = True
+
+    async def _handle_user_stopped_speaking(self, frame: UserStoppedSpeakingFrame):
+        if frame.emulated:
+            return
+
+        self._user_speaking = False
+
+        content = io.BytesIO()
+        wav = wave.open(content, "wb")
+        wav.setsampwidth(2)
+        wav.setnchannels(1)
+        wav.setframerate(self.sample_rate)
+        wav.writeframes(self._audio_buffer)
+        wav.close()
+        content.seek(0)
+
+        await self.process_generator(self.run_stt(content.read()))
+
+        # Start clean.
+        self._audio_buffer.clear()
+
+    async def process_audio_frame(self, frame: AudioRawFrame, direction: FrameDirection):
+        # If the user is speaking the audio buffer will keep growing.
+        self._audio_buffer += frame.audio
+
+        # If the user is not speaking we keep just a little bit of audio.
+        if not self._user_speaking and len(self._audio_buffer) > self._audio_buffer_size_1s:
+            discarded = len(self._audio_buffer) - self._audio_buffer_size_1s
+            self._audio_buffer = self._audio_buffer[discarded:]

src/pipecat/services/tavus/video.py

@@ -23,7 +23,7 @@ from pipecat.frames.frames import (
     TTSStoppedFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
-from pipecat.services.ai_services import AIService
+from pipecat.services.ai_service import AIService
 
 
 class TavusVideoService(AIService):

src/pipecat/services/tts_service.py

@@ -0,0 +1,602 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+from abc import abstractmethod
+from typing import Any, AsyncGenerator, Dict, List, Mapping, Optional, Sequence, Tuple
+
+from loguru import logger
+
+from pipecat.frames.frames import (
+    BotStartedSpeakingFrame,
+    BotStoppedSpeakingFrame,
+    CancelFrame,
+    EndFrame,
+    ErrorFrame,
+    Frame,
+    InterimTranscriptionFrame,
+    LLMFullResponseEndFrame,
+    StartFrame,
+    StartInterruptionFrame,
+    TextFrame,
+    TranscriptionFrame,
+    TTSAudioRawFrame,
+    TTSSpeakFrame,
+    TTSStartedFrame,
+    TTSStoppedFrame,
+    TTSTextFrame,
+    TTSUpdateSettingsFrame,
+)
+from pipecat.processors.frame_processor import FrameDirection
+from pipecat.services.ai_service import AIService
+from pipecat.services.websocket_service import WebsocketService
+from pipecat.transcriptions.language import Language
+from pipecat.utils.text.base_text_aggregator import BaseTextAggregator
+from pipecat.utils.text.base_text_filter import BaseTextFilter
+from pipecat.utils.text.simple_text_aggregator import SimpleTextAggregator
+from pipecat.utils.time import seconds_to_nanoseconds
+
+
+class TTSService(AIService):
+    def __init__(
+        self,
+        *,
+        aggregate_sentences: bool = True,
+        # if True, TTSService will push TextFrames and LLMFullResponseEndFrames,
+        # otherwise subclass must do it
+        push_text_frames: bool = True,
+        # if True, TTSService will push TTSStoppedFrames, otherwise subclass must do it
+        push_stop_frames: bool = False,
+        # if push_stop_frames is True, wait for this idle period before pushing TTSStoppedFrame
+        stop_frame_timeout_s: float = 2.0,
+        # if True, TTSService will push silence audio frames after TTSStoppedFrame
+        push_silence_after_stop: bool = False,
+        # if push_silence_after_stop is True, send this amount of audio silence
+        silence_time_s: float = 2.0,
+        # if True, we will pause processing frames while we are receiving audio
+        pause_frame_processing: bool = False,
+        # TTS output sample rate
+        sample_rate: Optional[int] = None,
+        # Text aggregator to aggregate incoming tokens and decide when to push to the TTS.
+        text_aggregator: Optional[BaseTextAggregator] = None,
+        # Text filter executed after text has been aggregated.
+        text_filters: Sequence[BaseTextFilter] = [],
+        text_filter: Optional[BaseTextFilter] = None,
+        **kwargs,
+    ):
+        super().__init__(**kwargs)
+        self._aggregate_sentences: bool = aggregate_sentences
+        self._push_text_frames: bool = push_text_frames
+        self._push_stop_frames: bool = push_stop_frames
+        self._stop_frame_timeout_s: float = stop_frame_timeout_s
+        self._push_silence_after_stop: bool = push_silence_after_stop
+        self._silence_time_s: float = silence_time_s
+        self._pause_frame_processing: bool = pause_frame_processing
+        self._init_sample_rate = sample_rate
+        self._sample_rate = 0
+        self._voice_id: str = ""
+        self._settings: Dict[str, Any] = {}
+        self._text_aggregator: BaseTextAggregator = text_aggregator or SimpleTextAggregator()
+        self._text_filters: Sequence[BaseTextFilter] = text_filters
+        if text_filter:
+            import warnings
+
+            with warnings.catch_warnings():
+                warnings.simplefilter("always")
+                warnings.warn(
+                    "Parameter 'text_filter' is deprecated, use 'text_filters' instead.",
+                    DeprecationWarning,
+                )
+            self._text_filters = [text_filter]
+
+        self._stop_frame_task: Optional[asyncio.Task] = None
+        self._stop_frame_queue: asyncio.Queue = asyncio.Queue()
+
+        self._processing_text: bool = False
+
+    @property
+    def sample_rate(self) -> int:
+        return self._sample_rate
+
+    async def set_model(self, model: str):
+        self.set_model_name(model)
+
+    def set_voice(self, voice: str):
+        self._voice_id = voice
+
+    # Converts the text to audio.
+    @abstractmethod
+    async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
+        pass
+
+    def language_to_service_language(self, language: Language) -> Optional[str]:
+        return Language(language)
+
+    async def update_setting(self, key: str, value: Any):
+        pass
+
+    async def flush_audio(self):
+        pass
+
+    async def start(self, frame: StartFrame):
+        await super().start(frame)
+        self._sample_rate = self._init_sample_rate or frame.audio_out_sample_rate
+        if self._push_stop_frames and not self._stop_frame_task:
+            self._stop_frame_task = self.create_task(self._stop_frame_handler())
+
+    async def stop(self, frame: EndFrame):
+        await super().stop(frame)
+        if self._stop_frame_task:
+            await self.cancel_task(self._stop_frame_task)
+            self._stop_frame_task = None
+
+    async def cancel(self, frame: CancelFrame):
+        await super().cancel(frame)
+        if self._stop_frame_task:
+            await self.cancel_task(self._stop_frame_task)
+            self._stop_frame_task = None
+
+    async def _update_settings(self, settings: Mapping[str, Any]):
+        for key, value in settings.items():
+            if key in self._settings:
+                logger.info(f"Updating TTS setting {key} to: [{value}]")
+                self._settings[key] = value
+                if key == "language":
+                    self._settings[key] = self.language_to_service_language(value)
+            elif key == "model":
+                self.set_model_name(value)
+            elif key == "voice":
+                self.set_voice(value)
+            elif key == "text_filter":
+                for filter in self._text_filters:
+                    filter.update_settings(value)
+            else:
+                logger.warning(f"Unknown setting for TTS service: {key}")
+
+    async def say(self, text: str):
+        await self.queue_frame(TTSSpeakFrame(text))
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if (
+            isinstance(frame, TextFrame)
+            and not isinstance(frame, InterimTranscriptionFrame)
+            and not isinstance(frame, TranscriptionFrame)
+        ):
+            await self._process_text_frame(frame)
+        elif isinstance(frame, StartInterruptionFrame):
+            await self._handle_interruption(frame, direction)
+            await self.push_frame(frame, direction)
+        elif isinstance(frame, (LLMFullResponseEndFrame, EndFrame)):
+            # We pause processing incoming frames if the LLM response included
+            # text (it might be that it's only a function calling response). We
+            # pause to avoid audio overlapping.
+            await self._maybe_pause_frame_processing()
+
+            sentence = self._text_aggregator.text
+            self._text_aggregator.reset()
+            self._processing_text = False
+            await self._push_tts_frames(sentence)
+            if isinstance(frame, LLMFullResponseEndFrame):
+                if self._push_text_frames:
+                    await self.push_frame(frame, direction)
+            else:
+                await self.push_frame(frame, direction)
+        elif isinstance(frame, TTSSpeakFrame):
+            # Store if we were processing text or not so we can set it back.
+            processing_text = self._processing_text
+            await self._push_tts_frames(frame.text)
+            # We pause processing incoming frames because we are sending data to
+            # the TTS. We pause to avoid audio overlapping.
+            await self._maybe_pause_frame_processing()
+            await self.flush_audio()
+            self._processing_text = processing_text
+        elif isinstance(frame, TTSUpdateSettingsFrame):
+            await self._update_settings(frame.settings)
+        elif isinstance(frame, BotStoppedSpeakingFrame):
+            await self._maybe_resume_frame_processing()
+            await self.push_frame(frame, direction)
+        else:
+            await self.push_frame(frame, direction)
+
+    async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
+        if self._push_silence_after_stop and isinstance(frame, TTSStoppedFrame):
+            silence_num_bytes = int(self._silence_time_s * self.sample_rate * 2)  # 16-bit
+            await self.push_frame(
+                TTSAudioRawFrame(
+                    audio=b"\x00" * silence_num_bytes,
+                    sample_rate=self.sample_rate,
+                    num_channels=1,
+                )
+            )
+
+        await super().push_frame(frame, direction)
+
+        if self._push_stop_frames and (
+            isinstance(frame, StartInterruptionFrame)
+            or isinstance(frame, TTSStartedFrame)
+            or isinstance(frame, TTSAudioRawFrame)
+            or isinstance(frame, TTSStoppedFrame)
+        ):
+            await self._stop_frame_queue.put(frame)
+
+    async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection):
+        self._processing_text = False
+        self._text_aggregator.handle_interruption()
+        for filter in self._text_filters:
+            filter.handle_interruption()
+
+    async def _maybe_pause_frame_processing(self):
+        if self._processing_text and self._pause_frame_processing:
+            await self.pause_processing_frames()
+
+    async def _maybe_resume_frame_processing(self):
+        if self._pause_frame_processing:
+            await self.resume_processing_frames()
+
+    async def _process_text_frame(self, frame: TextFrame):
+        text: Optional[str] = None
+        if not self._aggregate_sentences:
+            text = frame.text
+        else:
+            text = self._text_aggregator.aggregate(frame.text)
+
+        if text:
+            await self._push_tts_frames(text)
+
+    async def _push_tts_frames(self, text: str):
+        # Remove leading newlines only
+        text = text.lstrip("\n")
+
+        # Don't send only whitespace. This causes problems for some TTS models. But also don't
+        # strip all whitespace, as whitespace can influence prosody.
+        if not text.strip():
+            return
+
+        # This is just a flag that indicates if we sent something to the TTS
+        # service. It will be cleared if we sent text because of a TTSSpeakFrame
+        # or when we received an LLMFullResponseEndFrame
+        self._processing_text = True
+
+        await self.start_processing_metrics()
+
+        # Process all filter.
+        for filter in self._text_filters:
+            filter.reset_interruption()
+            text = filter.filter(text)
+
+        await self.process_generator(self.run_tts(text))
+
+        await self.stop_processing_metrics()
+
+        if self._push_text_frames:
+            # We send the original text after the audio. This way, if we are
+            # interrupted, the text is not added to the assistant context.
+            await self.push_frame(TTSTextFrame(text))
+
+    async def _stop_frame_handler(self):
+        has_started = False
+        while True:
+            try:
+                frame = await asyncio.wait_for(
+                    self._stop_frame_queue.get(), self._stop_frame_timeout_s
+                )
+                if isinstance(frame, TTSStartedFrame):
+                    has_started = True
+                elif isinstance(frame, (TTSStoppedFrame, StartInterruptionFrame)):
+                    has_started = False
+            except asyncio.TimeoutError:
+                if has_started:
+                    await self.push_frame(TTSStoppedFrame())
+                    has_started = False
+
+
+class WordTTSService(TTSService):
+    """This is a base class for TTS services that support word timestamps. Word
+    timestamps are useful to synchronize audio with text of the spoken
+    words. This way only the spoken words are added to the conversation context.
+
+    """
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self._initial_word_timestamp = -1
+        self._words_queue = asyncio.Queue()
+        self._words_task = None
+
+    def start_word_timestamps(self):
+        if self._initial_word_timestamp == -1:
+            self._initial_word_timestamp = self.get_clock().get_time()
+
+    def reset_word_timestamps(self):
+        self._initial_word_timestamp = -1
+
+    async def add_word_timestamps(self, word_times: List[Tuple[str, float]]):
+        for word, timestamp in word_times:
+            await self._words_queue.put((word, seconds_to_nanoseconds(timestamp)))
+
+    async def start(self, frame: StartFrame):
+        await super().start(frame)
+        self._create_words_task()
+
+    async def stop(self, frame: EndFrame):
+        await super().stop(frame)
+        await self._stop_words_task()
+
+    async def cancel(self, frame: CancelFrame):
+        await super().cancel(frame)
+        await self._stop_words_task()
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, (LLMFullResponseEndFrame, EndFrame)):
+            await self.flush_audio()
+
+    async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection):
+        await super()._handle_interruption(frame, direction)
+        self.reset_word_timestamps()
+
+    def _create_words_task(self):
+        if not self._words_task:
+            self._words_task = self.create_task(self._words_task_handler())
+
+    async def _stop_words_task(self):
+        if self._words_task:
+            await self.cancel_task(self._words_task)
+            self._words_task = None
+
+    async def _words_task_handler(self):
+        last_pts = 0
+        while True:
+            (word, timestamp) = await self._words_queue.get()
+            if word == "Reset" and timestamp == 0:
+                self.reset_word_timestamps()
+                frame = None
+            elif word == "LLMFullResponseEndFrame" and timestamp == 0:
+                frame = LLMFullResponseEndFrame()
+                frame.pts = last_pts
+            elif word == "TTSStoppedFrame" and timestamp == 0:
+                frame = TTSStoppedFrame()
+                frame.pts = last_pts
+            else:
+                frame = TTSTextFrame(word)
+                frame.pts = self._initial_word_timestamp + timestamp
+            if frame:
+                last_pts = frame.pts
+                await self.push_frame(frame)
+            self._words_queue.task_done()
+
+
+class WebsocketTTSService(TTSService, WebsocketService):
+    """This is a base class for websocket-based TTS services.
+
+    If an error occurs with the websocket, an "on_connection_error" event will
+    be triggered:
+
+       @tts.event_handler("on_connection_error")
+       async def on_connection_error(tts: TTSService, error: str):
+           ...
+
+    """
+
+    def __init__(self, *, reconnect_on_error: bool = True, **kwargs):
+        TTSService.__init__(self, **kwargs)
+        WebsocketService.__init__(self, reconnect_on_error=reconnect_on_error, **kwargs)
+        self._register_event_handler("on_connection_error")
+
+    async def _report_error(self, error: ErrorFrame):
+        await self._call_event_handler("on_connection_error", error.error)
+        await self.push_error(error)
+
+
+class InterruptibleTTSService(WebsocketTTSService):
+    """This is a base class for websocket-based TTS services that don't support
+    word timestamps and that don't offer a way to correlate the generated audio
+    to the requested text.
+
+    """
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+        # Indicates if the bot is speaking. If the bot is not speaking we don't
+        # need to reconnect when the user speaks. If the bot is speaking and the
+        # user interrupts we need to reconnect.
+        self._bot_speaking = False
+
+    async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection):
+        await super()._handle_interruption(frame, direction)
+        if self._bot_speaking:
+            await self._disconnect()
+            await self._connect()
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, BotStartedSpeakingFrame):
+            self._bot_speaking = True
+        elif isinstance(frame, BotStoppedSpeakingFrame):
+            self._bot_speaking = False
+
+
+class WebsocketWordTTSService(WordTTSService, WebsocketService):
+    """This is a base class for websocket-based TTS services that support word
+    timestamps.
+
+    If an error occurs with the websocket a "on_connection_error" event will be
+    triggered:
+
+       @tts.event_handler("on_connection_error")
+       async def on_connection_error(tts: TTSService, error: str):
+           ...
+
+    """
+
+    def __init__(self, *, reconnect_on_error: bool = True, **kwargs):
+        WordTTSService.__init__(self, **kwargs)
+        WebsocketService.__init__(self, reconnect_on_error=reconnect_on_error, **kwargs)
+        self._register_event_handler("on_connection_error")
+
+    async def _report_error(self, error: ErrorFrame):
+        await self._call_event_handler("on_connection_error", error.error)
+        await self.push_error(error)
+
+
+class InterruptibleWordTTSService(WebsocketWordTTSService):
+    """This is a base class for websocket-based TTS services that support word
+    timestamps but don't offer a way to correlate the generated audio to the
+    requested text.
+
+    """
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+        # Indicates if the bot is speaking. If the bot is not speaking we don't
+        # need to reconnect when the user speaks. If the bot is speaking and the
+        # user interrupts we need to reconnect.
+        self._bot_speaking = False
+
+    async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection):
+        await super()._handle_interruption(frame, direction)
+        if self._bot_speaking:
+            await self._disconnect()
+            await self._connect()
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, BotStartedSpeakingFrame):
+            self._bot_speaking = True
+        elif isinstance(frame, BotStoppedSpeakingFrame):
+            self._bot_speaking = False
+
+
+class AudioContextWordTTSService(WebsocketWordTTSService):
+    """This is a base class for websocket-based TTS services that support word
+    timestamps and also allow correlating the generated audio with the requested
+    text.
+
+    Each request could be multiple sentences long which are grouped by
+    context. For this to work, the TTS service needs to support handling
+    multiple requests at once (i.e. multiple simultaneous contexts).
+
+    The audio received from the TTS will be played in context order. That is, if
+    we requested audio for a context "A" and then audio for context "B", the
+    audio from context ID "A" will be played first.
+
+    """
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self._contexts_queue = asyncio.Queue()
+        self._contexts: Dict[str, asyncio.Queue] = {}
+        self._audio_context_task = None
+
+    async def create_audio_context(self, context_id: str):
+        """Create a new audio context."""
+        await self._contexts_queue.put(context_id)
+        self._contexts[context_id] = asyncio.Queue()
+        logger.trace(f"{self} created audio context {context_id}")
+
+    async def append_to_audio_context(self, context_id: str, frame: TTSAudioRawFrame):
+        """Append audio to an existing context."""
+        if self.audio_context_available(context_id):
+            logger.trace(f"{self} appending audio {frame} to audio context {context_id}")
+            await self._contexts[context_id].put(frame)
+        else:
+            logger.warning(f"{self} unable to append audio to context {context_id}")
+
+    async def remove_audio_context(self, context_id: str):
+        """Remove an existing audio context."""
+        if self.audio_context_available(context_id):
+            # We just mark the audio context for deletion by appending
+            # None. Once we reach None while handling audio we know we can
+            # safely remove the context.
+            logger.trace(f"{self} marking audio context {context_id} for deletion")
+            await self._contexts[context_id].put(None)
+        else:
+            logger.warning(f"{self} unable to remove context {context_id}")
+
+    def audio_context_available(self, context_id: str) -> bool:
+        """Checks whether the given audio context is registered."""
+        return context_id in self._contexts
+
+    async def start(self, frame: StartFrame):
+        await super().start(frame)
+        self._create_audio_context_task()
+
+    async def stop(self, frame: EndFrame):
+        await super().stop(frame)
+        if self._audio_context_task:
+            # Indicate no more audio contexts are available. this will end the
+            # task cleanly after all contexts have been processed.
+            await self._contexts_queue.put(None)
+            await self.wait_for_task(self._audio_context_task)
+            self._audio_context_task = None
+
+    async def cancel(self, frame: CancelFrame):
+        await super().cancel(frame)
+        await self._stop_audio_context_task()
+
+    async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection):
+        await super()._handle_interruption(frame, direction)
+        await self._stop_audio_context_task()
+        self._create_audio_context_task()
+
+    def _create_audio_context_task(self):
+        if not self._audio_context_task:
+            self._contexts_queue = asyncio.Queue()
+            self._contexts: Dict[str, asyncio.Queue] = {}
+            self._audio_context_task = self.create_task(self._audio_context_task_handler())
+
+    async def _stop_audio_context_task(self):
+        if self._audio_context_task:
+            await self.cancel_task(self._audio_context_task)
+            self._audio_context_task = None
+
+    async def _audio_context_task_handler(self):
+        """In this task we process audio contexts in order."""
+        running = True
+        while running:
+            context_id = await self._contexts_queue.get()
+
+            if context_id:
+                # Process the audio context until the context doesn't have more
+                # audio available (i.e. we find None).
+                await self._handle_audio_context(context_id)
+
+                # We just finished processing the context, so we can safely remove it.
+                del self._contexts[context_id]
+
+                # Append some silence between sentences.
+                silence = b"\x00" * self.sample_rate
+                frame = TTSAudioRawFrame(
+                    audio=silence, sample_rate=self.sample_rate, num_channels=1
+                )
+                await self.push_frame(frame)
+            else:
+                running = False
+
+            self._contexts_queue.task_done()
+
+    async def _handle_audio_context(self, context_id: str):
+        # If we don't receive any audio during this time, we consider the context finished.
+        AUDIO_CONTEXT_TIMEOUT = 3.0
+        queue = self._contexts[context_id]
+        running = True
+        while running:
+            try:
+                frame = await asyncio.wait_for(queue.get(), timeout=AUDIO_CONTEXT_TIMEOUT)
+                if frame:
+                    await self.push_frame(frame)
+                running = frame is not None
+            except asyncio.TimeoutError:
+                # We didn't get audio, so let's consider this context finished.
+                logger.trace(f"{self} time out on audio context {context_id}")
+                break

src/pipecat/services/ultravox/stt.py

@@ -29,7 +29,7 @@ from pipecat.frames.frames import (
     UserStoppedSpeakingFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
-from pipecat.services.ai_services import AIService
+from pipecat.services.ai_service import AIService
 
 try:
     from transformers import AutoTokenizer
@@ -177,7 +177,7 @@ class UltravoxSTTService(AIService):
     to generate text transcriptions.
 
     Args:
-        model_size: The Ultravox model to use (ModelSize enum or string)
+        model_name: The Ultravox model to use (ModelSize enum or string)
         hf_token: Hugging Face token for model access
         temperature: Sampling temperature for generation
         max_tokens: Maximum tokens to generate
@@ -194,7 +194,7 @@ class UltravoxSTTService(AIService):
     def __init__(
         self,
         *,
-        model_size: str = "fixie-ai/ultravox-v0_4_1-llama-3_1-8b",
+        model_name: str = "fixie-ai/ultravox-v0_5-llama-3_1-8b",
         hf_token: Optional[str] = None,
         temperature: float = 0.7,
         max_tokens: int = 100,
@@ -211,7 +211,6 @@ class UltravoxSTTService(AIService):
             logger.warning("No Hugging Face token provided. Model may not load correctly.")
 
         # Initialize model
-        model_name = model_size if isinstance(model_size, str) else model_size.value
         self._model = UltravoxModel(model_name=model_name)
 
         # Initialize service state
@@ -356,10 +355,10 @@ class UltravoxSTTService(AIService):
                     await self.start_ttfb_metrics()
                     await self.start_processing_metrics()
 
-                    async for response in self.model.generate(
+                    async for response in self._model.generate(
                         messages=[{"role": "user", "content": "<|audio|>\n"}],
-                        temperature=self.temperature,
-                        max_tokens=self.max_tokens,
+                        temperature=self._temperature,
+                        max_tokens=self._max_tokens,
                         audio=audio_float32,
                     ):
                         # Stop TTFB metrics after first response

src/pipecat/services/vision_service.py

@@ -0,0 +1,34 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+from abc import abstractmethod
+from typing import AsyncGenerator
+
+from pipecat.frames.frames import Frame, VisionImageRawFrame
+from pipecat.processors.frame_processor import FrameDirection
+from pipecat.services.ai_service import AIService
+
+
+class VisionService(AIService):
+    """VisionService is a base class for vision services."""
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self._describe_text = None
+
+    @abstractmethod
+    async def run_vision(self, frame: VisionImageRawFrame) -> AsyncGenerator[Frame, None]:
+        pass
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, VisionImageRawFrame):
+            await self.start_processing_metrics()
+            await self.process_generator(self.run_vision(frame))
+            await self.stop_processing_metrics()
+        else:
+            await self.push_frame(frame, direction)

src/pipecat/services/whisper/base_stt.py

@@ -11,7 +11,7 @@ from openai import AsyncOpenAI
 from openai.types.audio import Transcription
 
 from pipecat.frames.frames import ErrorFrame, Frame, TranscriptionFrame
-from pipecat.services.ai_services import SegmentedSTTService
+from pipecat.services.stt_service import SegmentedSTTService
 from pipecat.transcriptions.language import Language
 from pipecat.utils.time import time_now_iso8601
 

src/pipecat/services/whisper/stt.py

@@ -15,7 +15,7 @@ from loguru import logger
 from typing_extensions import TYPE_CHECKING, override
 
 from pipecat.frames.frames import ErrorFrame, Frame, TranscriptionFrame
-from pipecat.services.ai_services import SegmentedSTTService
+from pipecat.services.stt_service import SegmentedSTTService
 from pipecat.transcriptions.language import Language
 from pipecat.utils.time import time_now_iso8601
 

src/pipecat/services/xtts/tts.py

@@ -18,7 +18,7 @@ from pipecat.frames.frames import (
     TTSStartedFrame,
     TTSStoppedFrame,
 )
-from pipecat.services.ai_services import TTSService
+from pipecat.services.tts_service import TTSService
 from pipecat.transcriptions.language import Language
 
 # The server below can connect to XTTS through a local running docker

src/pipecat/transcriptions/language.py

@@ -182,6 +182,9 @@ class Language(StrEnum):
     GA = "ga"
     GA_IE = "ga-IE"
 
+    # Gaelic
+    GD = "gd"
+
     # Galician
     GL = "gl"
     GL_ES = "gl-ES"
@@ -193,6 +196,9 @@ class Language(StrEnum):
     # Hausa
     HA = "ha"
 
+    # Hawaiian
+    HAW = "haw"
+
     # Hebrew
     HE = "he"
     HE_IL = "he-IL"
@@ -288,6 +294,9 @@ class Language(StrEnum):
     # Malagasy
     MG = "mg"
 
+    # Maori
+    MI = "mi"
+
     # Macedonian
     MK = "mk"
     MK_MK = "mk-MK"
@@ -300,9 +309,6 @@ class Language(StrEnum):
     MN = "mn"
     MN_MN = "mn-MN"
 
-    # Maori
-    MI = "mi"
-
     # Marathi
     MR = "mr"
     MR_IN = "mr-IN"
@@ -318,6 +324,7 @@ class Language(StrEnum):
     # Burmese
     MY = "my"
     MY_MM = "my-MM"
+    MY_MR = "mymr"
 
     # Norwegian
     NB = "nb"  # Norwegian Bokmål
@@ -414,9 +421,6 @@ class Language(StrEnum):
     SW_KE = "sw-KE"
     SW_TZ = "sw-TZ"
 
-    # Tagalog
-    TL = "tl"
-
     # Tamil
     TA = "ta"
     TA_IN = "ta-IN"
@@ -438,6 +442,9 @@ class Language(StrEnum):
     # Turkmen
     TK = "tk"
 
+    # Tagalog
+    TL = "tl"
+
     # Turkish
     TR = "tr"
     TR_TR = "tr-TR"
@@ -489,7 +496,7 @@ class Language(StrEnum):
     ZH_TW = "zh-TW"
 
     # Xhosa
-    XH = "xh"
+    XH = "xh-ZA"
 
     # Zulu
     ZU = "zu"

src/pipecat/transports/base_output.py

@@ -79,10 +79,11 @@ class BaseOutputTransport(FrameProcessor):
     async def start(self, frame: StartFrame):
         self._sample_rate = self._params.audio_out_sample_rate or frame.audio_out_sample_rate
 
-        # We will write 20ms audio at a time. If we receive long audio frames we
+        # We will write 10ms*CHUNKS of audio at a time (where CHUNKS is the
+        # `audio_out_10ms_chunks` parameter). If we receive long audio frames we
         # will chunk them. This will help with interruption handling.
         audio_bytes_10ms = int(self._sample_rate / 100) * self._params.audio_out_channels * 2
-        self._audio_chunk_size = audio_bytes_10ms * 2
+        self._audio_chunk_size = audio_bytes_10ms * self._params.audio_out_10ms_chunks
 
         # Start audio mixer.
         if self._params.audio_out_mixer:
@@ -335,13 +336,22 @@ class BaseOutputTransport(FrameProcessor):
             return without_mixer(BOT_VAD_STOP_SECS)
 
     async def _sink_task_handler(self):
+        # Push a BotSpeakingFrame every 200ms, we don't really need to push it
+        # at every audio chunk. If the audio chunk is bigger than 200ms, push at
+        # every audio chunk.
+        TOTAL_CHUNK_MS = self._params.audio_out_10ms_chunks * 10
+        BOT_SPEAKING_CHUNK_PERIOD = max(int(200 / TOTAL_CHUNK_MS), 1)
+        bot_speaking_counter = 0
         async for frame in self._next_frame():
             # Notify the bot started speaking upstream if necessary and that
             # it's actually speaking.
             if isinstance(frame, TTSAudioRawFrame):
                 await self._bot_started_speaking()
-                await self.push_frame(BotSpeakingFrame())
-                await self.push_frame(BotSpeakingFrame(), FrameDirection.UPSTREAM)
+                if bot_speaking_counter % BOT_SPEAKING_CHUNK_PERIOD == 0:
+                    await self.push_frame(BotSpeakingFrame())
+                    await self.push_frame(BotSpeakingFrame(), FrameDirection.UPSTREAM)
+                    bot_speaking_counter = 0
+                bot_speaking_counter += 1
 
             # No need to push EndFrame, it's pushed from process_frame().
             if isinstance(frame, EndFrame):

src/pipecat/transports/base_transport.py

@@ -31,6 +31,7 @@ class TransportParams(BaseModel):
     audio_out_sample_rate: Optional[int] = None
     audio_out_channels: int = 1
     audio_out_bitrate: int = 96000
+    audio_out_10ms_chunks: int = 4
     audio_out_mixer: Optional[BaseAudioMixer] = None
     audio_in_enabled: bool = False
     audio_in_sample_rate: Optional[int] = None

src/pipecat/transports/network/fastapi_websocket.py

@@ -61,6 +61,10 @@ class FastAPIWebsocketClient:
         self._closing = False
         self._is_binary = is_binary
         self._callbacks = callbacks
+        self._leave_counter = 0
+
+    async def setup(self, _: StartFrame):
+        self._leave_counter += 1
 
     def receive(self) -> typing.AsyncIterator[bytes | str]:
         return self._websocket.iter_bytes() if self._is_binary else self._websocket.iter_text()
@@ -73,6 +77,10 @@ class FastAPIWebsocketClient:
                 await self._websocket.send_text(data)
 
     async def disconnect(self):
+        self._leave_counter -= 1
+        if self._leave_counter > 0:
+            return
+
         if self.is_connected and not self.is_closing:
             self._closing = True
             await self._websocket.close()
@@ -116,6 +124,7 @@ class FastAPIWebsocketInputTransport(BaseInputTransport):
 
     async def start(self, frame: StartFrame):
         await super().start(frame)
+        await self._client.setup(frame)
         await self._params.serializer.setup(frame)
         if not self._monitor_websocket_task and self._params.session_timeout:
             self._monitor_websocket_task = self.create_task(self._monitor_websocket())
@@ -192,6 +201,7 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
 
     async def start(self, frame: StartFrame):
         await super().start(frame)
+        await self._client.setup(frame)
         await self._params.serializer.setup(frame)
         self._send_interval = (self._audio_chunk_size / self.sample_rate) / 2
 

src/pipecat/transports/network/small_webrtc.py

@@ -138,6 +138,13 @@ class RawVideoTrack(VideoStreamTrack):
 
 
 class SmallWebRTCClient:
+    FORMAT_CONVERSIONS = {
+        "yuv420p": cv2.COLOR_YUV2RGB_I420,
+        "yuvj420p": cv2.COLOR_YUV2RGB_I420,  # OpenCV treats both the same
+        "nv12": cv2.COLOR_YUV2RGB_NV12,
+        "gray": cv2.COLOR_GRAY2RGB,
+    }
+
     def __init__(self, webrtc_connection: SmallWebRTCConnection, callbacks: SmallWebRTCCallbacks):
         self._webrtc_connection = webrtc_connection
         self._closing = False
@@ -176,6 +183,30 @@ class SmallWebRTCClient:
         async def on_app_message(connection: SmallWebRTCConnection, message: Any):
             await self._handle_app_message(message)
 
+    def _convert_frame(self, frame_array: np.ndarray, format_name: str) -> np.ndarray:
+        """
+        Convert a given frame to RGB format based on the input format.
+
+        Args:
+            frame_array (np.ndarray): The input frame.
+            format_name (str): The format of the input frame.
+
+        Returns:
+            np.ndarray: The converted RGB frame.
+
+        Raises:
+            ValueError: If the format is unsupported.
+        """
+        if format_name.startswith("rgb"):  # Already in RGB, no conversion needed
+            return frame_array
+
+        conversion_code = SmallWebRTCClient.FORMAT_CONVERSIONS.get(format_name)
+
+        if conversion_code is None:
+            raise ValueError(f"Unsupported format: {format_name}")
+
+        return cv2.cvtColor(frame_array, conversion_code)
+
     async def read_video_frame(self):
         """
         Reads a video frame from the given MediaStreamTrack, converts it to RGB,
@@ -203,21 +234,9 @@ class SmallWebRTCClient:
                 continue
 
             format_name = frame.format.name
-
             # Convert frame to NumPy array in its native format
             frame_array = frame.to_ndarray(format=format_name)
-
-            # Handle different formats dynamically
-            if format_name == "yuv420p":
-                frame_rgb = cv2.cvtColor(frame_array, cv2.COLOR_YUV2RGB_I420)
-            elif format_name == "nv12":
-                frame_rgb = cv2.cvtColor(frame_array, cv2.COLOR_YUV2RGB_NV12)
-            elif format_name == "gray":
-                frame_rgb = cv2.cvtColor(frame_array, cv2.COLOR_GRAY2RGB)
-            elif format_name.startswith("rgb"):  # Already RGB, no conversion needed
-                frame_rgb = frame_array
-            else:
-                raise ValueError(f"Unsupported format: {format_name}")
+            frame_rgb = self._convert_frame(frame_array, format_name)
 
             image_frame = InputImageRawFrame(
                 image=frame_rgb.tobytes(),

src/pipecat/transports/network/websocket_client.py

@@ -56,8 +56,8 @@ class WebsocketClientSession:
         self._callbacks = callbacks
         self._transport_name = transport_name
 
+        self._leave_counter = 0
         self._task_manager: Optional[BaseTaskManager] = None
-
         self._websocket: Optional[websockets.WebSocketClientProtocol] = None
 
     @property
@@ -69,6 +69,7 @@ class WebsocketClientSession:
         return self._task_manager
 
     async def setup(self, frame: StartFrame):
+        self._leave_counter += 1
         if not self._task_manager:
             self._task_manager = frame.task_manager
 
@@ -87,7 +88,8 @@ class WebsocketClientSession:
             logger.error(f"Timeout connecting to {self._uri}")
 
     async def disconnect(self):
-        if not self._websocket:
+        self._leave_counter -= 1
+        if not self._websocket or self._leave_counter > 0:
             return
 
         await self.task_manager.cancel_task(self._client_task)