Merge branch 'main' into sarvam/stt

src/pipecat/adapters/services/anthropic_adapter.py

@@ -245,13 +245,25 @@ class AnthropicLLMAdapter(BaseLLMAdapter[AnthropicLLMInvocationParams]):
                     item["text"] = "(empty)"
                 # handle image_url -> image conversion
                 if item["type"] == "image_url":
-                    item["type"] = "image"
-                    item["source"] = {
-                        "type": "base64",
-                        "media_type": "image/jpeg",
-                        "data": item["image_url"]["url"].split(",")[1],
-                    }
-                    del item["image_url"]
+                    if item["image_url"]["url"].startswith("data:"):
+                        item["type"] = "image"
+                        item["source"] = {
+                            "type": "base64",
+                            "media_type": "image/jpeg",
+                            "data": item["image_url"]["url"].split(",")[1],
+                        }
+                        del item["image_url"]
+                    elif item["image_url"]["url"].startswith("http"):
+                        item["type"] = "image"
+                        item["source"] = {
+                            "type": "url",
+                            "url": item["image_url"]["url"],
+                        }
+                        del item["image_url"]
+                    else:
+                        url = item["image_url"]["url"]
+                        logger.warning(f"Unsupported 'image_url': {url}")
+
             # In the case where there's a single image in the list (like what
             # would result from a UserImageRawFrame), ensure that the image
             # comes before text, as recommended by Anthropic docs

src/pipecat/adapters/services/bedrock_adapter.py

@@ -256,15 +256,22 @@ class AWSBedrockLLMAdapter(BaseLLMAdapter[AWSBedrockLLMInvocationParams]):
                     new_content.append({"text": text_content})
                 # handle image_url -> image conversion
                 if item["type"] == "image_url":
-                    new_item = {
-                        "image": {
-                            "format": "jpeg",
-                            "source": {
-                                "bytes": base64.b64decode(item["image_url"]["url"].split(",")[1])
-                            },
+                    if item["image_url"]["url"].startswith("data:"):
+                        new_item = {
+                            "image": {
+                                "format": "jpeg",
+                                "source": {
+                                    "bytes": base64.b64decode(
+                                        item["image_url"]["url"].split(",")[1]
+                                    )
+                                },
+                            }
                         }
-                    }
-                    new_content.append(new_item)
+                        new_content.append(new_item)
+                    else:
+                        url = item["image_url"]["url"]
+                        logger.warning(f"Unsupported 'image_url': {url}")
+
             # In the case where there's a single image in the list (like what
             # would result from a UserImageRawFrame), ensure that the image
             # comes before text

src/pipecat/adapters/services/gemini_adapter.py

@@ -343,7 +343,7 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
             for c in content:
                 if c["type"] == "text":
                     parts.append(Part(text=c["text"]))
-                elif c["type"] == "image_url":
+                elif c["type"] == "image_url" and c["image_url"]["url"].startswith("data:"):
                     parts.append(
                         Part(
                             inline_data=Blob(
@@ -352,6 +352,9 @@ class GeminiLLMAdapter(BaseLLMAdapter[GeminiLLMInvocationParams]):
                             )
                         )
                     )
+                elif c["type"] == "image_url":
+                    url = c["image_url"]["url"]
+                    logger.warning(f"Unsupported 'image_url': {url}")
                 elif c["type"] == "input_audio":
                     input_audio = c["input_audio"]
                     audio_bytes = base64.b64decode(input_audio["data"])

src/pipecat/frames/frames.py

@@ -1201,26 +1201,23 @@ class TransportMessageUrgentFrame(OutputTransportMessageUrgentFrame):
 class UserImageRequestFrame(SystemFrame):
     """Frame requesting an image from a specific user.
 
-    A frame to request an image from the given user. The frame might be
-    generated by a function call in which case the corresponding fields will be
-    properly set.
+    A frame to request an image from the given user. The request might come with
+    a text that can be later used to describe the requested image.
 
     Parameters:
         user_id: Identifier of the user to request image from.
-        context: Optional context for the image request.
-        function_name: Name of function that generated this request (if any).
-        tool_call_id: Tool call ID if generated by function call.
+        text: An optional text associated to the image request.
+        add_to_context: Whether the requested image should be added to an LLM context.
         video_source: Specific video source to capture from.
     """
 
     user_id: str
-    context: Optional[Any] = None
-    function_name: Optional[str] = None
-    tool_call_id: Optional[str] = None
+    text: Optional[str] = None
+    add_to_context: Optional[bool] = None
     video_source: Optional[str] = None
 
     def __str__(self):
-        return f"{self.name}(user: {self.user_id}, video_source: {self.video_source}, function: {self.function_name}, request: {self.tool_call_id})"
+        return f"{self.name}(user: {self.user_id}, text: {self.text}, add_to_context: {self.add_to_context}, {self.video_source})"
 
 
 @dataclass
@@ -1294,15 +1291,17 @@ class UserImageRawFrame(InputImageRawFrame):
 
     Parameters:
         user_id: Identifier of the user who provided this image.
-        request: The original image request frame if this is a response.
+        text: An optional text associated to this image.
+        add_to_context: Whether this image should be added to an LLM context.
     """
 
     user_id: str = ""
-    request: Optional[UserImageRequestFrame] = None
+    text: Optional[str] = None
+    add_to_context: Optional[bool] = None
 
     def __str__(self):
         pts = format_pts(self.pts)
-        return f"{self.name}(pts: {pts}, user: {self.user_id}, source: {self.transport_source}, size: {self.size}, format: {self.format}, request: {self.request})"
+        return f"{self.name}(pts: {pts}, user: {self.user_id}, source: {self.transport_source}, size: {self.size}, format: {self.format}, text: {self.text}, add_to_context: {self.add_to_context})"
 
 
 @dataclass

src/pipecat/processors/aggregators/llm_context.py

@@ -16,6 +16,7 @@ service-specific adapter.
 
 import base64
 import io
+import wave
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, List, Optional, TypeAlias, Union
 
@@ -113,6 +114,89 @@ class LLMContext:
         self._tools: ToolsSchema | NotGiven = LLMContext._normalize_and_validate_tools(tools)
         self._tool_choice: LLMContextToolChoice | NotGiven = tool_choice
 
+    @staticmethod
+    def create_image_url_message(
+        *,
+        role: str = "user",
+        url: str,
+        text: Optional[str] = None,
+    ) -> LLMContextMessage:
+        """Create a context message containing an image URL.
+
+        Args:
+            role: The role of this message (defaults to "user").
+            url: The URL of the image.
+            text: Optional text to include with the image.
+        """
+        content = []
+        if text:
+            content.append({"type": "text", "text": text})
+
+        content.append({"type": "image_url", "image_url": {"url": url}})
+
+        return {"role": role, "content": content}
+
+    @staticmethod
+    def create_image_message(
+        *,
+        role: str = "user",
+        format: str,
+        size: tuple[int, int],
+        image: bytes,
+        text: Optional[str] = None,
+    ) -> LLMContextMessage:
+        """Create a context message containing an image.
+
+        Args:
+            role: The role of this message (defaults to "user").
+            format: Image format (e.g., 'RGB', 'RGBA').
+            size: Image dimensions as (width, height) tuple.
+            image: Raw image bytes.
+            text: Optional text to include with the image.
+        """
+        buffer = io.BytesIO()
+        Image.frombytes(format, size, image).save(buffer, format="JPEG")
+        encoded_image = base64.b64encode(buffer.getvalue()).decode("utf-8")
+        url = f"data:image/jpeg;base64,{encoded_image}"
+
+        return LLMContext.create_image_url_message(role=role, url=url, text=text)
+
+    @staticmethod
+    def create_audio_message(
+        *, role: str = "user", audio_frames: list[AudioRawFrame], text: str = "Audio follows"
+    ) -> LLMContextMessage:
+        """Create a context message containing audio.
+
+        Args:
+            role: The role of this message (defaults to "user").
+            audio_frames: List of audio frame objects to include.
+            text: Optional text to include with the audio.
+        """
+        sample_rate = audio_frames[0].sample_rate
+        num_channels = audio_frames[0].num_channels
+
+        content = []
+        content.append({"type": "text", "text": text})
+        data = b"".join(frame.audio for frame in audio_frames)
+
+        with io.BytesIO() as buffer:
+            with wave.open(buffer, "wb") as wf:
+                wf.setsampwidth(2)
+                wf.setnchannels(num_channels)
+                wf.setframerate(sample_rate)
+                wf.writeframes(data)
+
+        encoded_audio = base64.b64encode(buffer.getvalue()).decode("utf-8")
+
+        content.append(
+            {
+                "type": "input_audio",
+                "input_audio": {"data": encoded_audio, "format": "wav"},
+            }
+        )
+
+        return {"role": role, "content": content}
+
     @property
     def messages(self) -> List[LLMContextMessage]:
         """Get the current messages list.
@@ -238,7 +322,7 @@ class LLMContext:
         self._tool_choice = tool_choice
 
     def add_image_frame_message(
-        self, *, format: str, size: tuple[int, int], image: bytes, text: str = None
+        self, *, format: str, size: tuple[int, int], image: bytes, text: Optional[str] = None
     ):
         """Add a message containing an image frame.
 
@@ -248,17 +332,8 @@ class LLMContext:
             image: Raw image bytes.
             text: Optional text to include with the image.
         """
-        buffer = io.BytesIO()
-        Image.frombytes(format, size, image).save(buffer, format="JPEG")
-        encoded_image = base64.b64encode(buffer.getvalue()).decode("utf-8")
-
-        content = []
-        if text:
-            content.append({"type": "text", "text": text})
-        content.append(
-            {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{encoded_image}"}},
-        )
-        self.add_message({"role": "user", "content": content})
+        message = LLMContext.create_image_message(format=format, size=size, image=image, text=text)
+        self.add_message(message)
 
     def add_audio_frames_message(
         self, *, audio_frames: list[AudioRawFrame], text: str = "Audio follows"
@@ -269,66 +344,8 @@ class LLMContext:
             audio_frames: List of audio frame objects to include.
             text: Optional text to include with the audio.
         """
-        if not audio_frames:
-            return
-
-        sample_rate = audio_frames[0].sample_rate
-        num_channels = audio_frames[0].num_channels
-
-        content = []
-        content.append({"type": "text", "text": text})
-        data = b"".join(frame.audio for frame in audio_frames)
-        data = bytes(
-            self._create_wav_header(
-                sample_rate,
-                num_channels,
-                16,
-                len(data),
-            )
-            + data
-        )
-        encoded_audio = base64.b64encode(data).decode("utf-8")
-        content.append(
-            {
-                "type": "input_audio",
-                "input_audio": {"data": encoded_audio, "format": "wav"},
-            }
-        )
-        self.add_message({"role": "user", "content": content})
-
-    def _create_wav_header(self, sample_rate, num_channels, bits_per_sample, data_size):
-        """Create a WAV file header for audio data.
-
-        Args:
-            sample_rate: Audio sample rate in Hz.
-            num_channels: Number of audio channels.
-            bits_per_sample: Bits per audio sample.
-            data_size: Size of audio data in bytes.
-
-        Returns:
-            WAV header as a bytearray.
-        """
-        # RIFF chunk descriptor
-        header = bytearray()
-        header.extend(b"RIFF")  # ChunkID
-        header.extend((data_size + 36).to_bytes(4, "little"))  # ChunkSize: total size - 8
-        header.extend(b"WAVE")  # Format
-        # "fmt " sub-chunk
-        header.extend(b"fmt ")  # Subchunk1ID
-        header.extend((16).to_bytes(4, "little"))  # Subchunk1Size (16 for PCM)
-        header.extend((1).to_bytes(2, "little"))  # AudioFormat (1 for PCM)
-        header.extend(num_channels.to_bytes(2, "little"))  # NumChannels
-        header.extend(sample_rate.to_bytes(4, "little"))  # SampleRate
-        # Calculate byte rate and block align
-        byte_rate = sample_rate * num_channels * (bits_per_sample // 8)
-        block_align = num_channels * (bits_per_sample // 8)
-        header.extend(byte_rate.to_bytes(4, "little"))  # ByteRate
-        header.extend(block_align.to_bytes(2, "little"))  # BlockAlign
-        header.extend(bits_per_sample.to_bytes(2, "little"))  # BitsPerSample
-        # "data" sub-chunk
-        header.extend(b"data")  # Subchunk2ID
-        header.extend(data_size.to_bytes(4, "little"))  # Subchunk2Size
-        return header
+        message = LLMContext.create_audio_message(audio_frames=audio_frames, text=text)
+        self.add_message(message)
 
     @staticmethod
     def _normalize_and_validate_tools(tools: ToolsSchema | NotGiven) -> ToolsSchema | NotGiven:

src/pipecat/processors/aggregators/llm_response_universal.py

@@ -616,7 +616,7 @@ class LLMAssistantAggregator(LLMContextAggregator):
             await self._handle_function_call_result(frame)
         elif isinstance(frame, FunctionCallCancelFrame):
             await self._handle_function_call_cancel(frame)
-        elif isinstance(frame, UserImageRawFrame) and frame.request and frame.request.tool_call_id:
+        elif isinstance(frame, UserImageRawFrame):
             await self._handle_user_image_frame(frame)
         elif isinstance(frame, BotStoppedSpeakingFrame):
             await self.push_aggregation()
@@ -767,27 +767,16 @@ class LLMAssistantAggregator(LLMContextAggregator):
                 message["content"] = result
 
     async def _handle_user_image_frame(self, frame: UserImageRawFrame):
-        logger.debug(
-            f"{self} UserImageRawFrame: [{frame.request.function_name}:{frame.request.tool_call_id}]"
-        )
-
-        if frame.request.tool_call_id not in self._function_calls_in_progress:
-            logger.warning(
-                f"UserImageRawFrame tool_call_id [{frame.request.tool_call_id}] is not running"
-            )
+        if not frame.add_to_context:
             return
 
-        del self._function_calls_in_progress[frame.request.tool_call_id]
+        logger.debug(f"{self} Adding UserImageRawFrame to LLM context (size: {frame.size})")
 
-        # Update context with the image frame
-        self._update_function_call_result(
-            frame.request.function_name, frame.request.tool_call_id, "COMPLETED"
-        )
         self._context.add_image_frame_message(
             format=frame.format,
             size=frame.size,
             image=frame.image,
-            text=frame.request.context,
+            text=frame.text,
         )
 
         await self.push_aggregation()

src/pipecat/processors/aggregators/user_response.py

@@ -27,11 +27,24 @@ class UserResponseAggregator(LLMUserAggregator):
     def __init__(self, **kwargs):
         """Initialize the user response aggregator.
 
+        .. deprecated:: 0.0.92
+            `UserResponseAggregator` is deprecated and will be removed in a future version.
+
         Args:
             **kwargs: Additional arguments passed to parent LLMUserAggregator.
         """
         super().__init__(context=LLMContext(), **kwargs)
 
+        import warnings
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "`UserResponseAggregator` is deprecated and will be removed in a future version.",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+
     async def push_aggregation(self):
         """Push the aggregated user response as a TextFrame.
 

src/pipecat/services/deepgram/flux/stt.py

@@ -156,6 +156,12 @@ class DeepgramFluxSTTService(WebsocketSTTService):
         self._language = Language.EN
         self._websocket_url = None
         self._receive_task = None
+        # Flux event handlers
+        self._register_event_handler("on_start_of_turn")
+        self._register_event_handler("on_turn_resumed")
+        self._register_event_handler("on_end_of_turn")
+        self._register_event_handler("on_eager_end_of_turn")
+        self._register_event_handler("on_update")
 
     async def _connect(self):
         """Connect to WebSocket and start background tasks.
@@ -523,6 +529,7 @@ class DeepgramFluxSTTService(WebsocketSTTService):
         await self.push_frame(UserStartedSpeakingFrame(), FrameDirection.DOWNSTREAM)
         await self.push_frame(UserStartedSpeakingFrame(), FrameDirection.UPSTREAM)
         await self.start_metrics()
+        await self._call_event_handler("on_start_of_turn", transcript)
         if transcript:
             logger.trace(f"Start of turn transcript: {transcript}")
 
@@ -537,6 +544,7 @@ class DeepgramFluxSTTService(WebsocketSTTService):
             event: The event type string for logging purposes.
         """
         logger.trace(f"Received event TurnResumed: {event}")
+        await self._call_event_handler("on_turn_resumed")
 
     async def _handle_end_of_turn(self, transcript: str, data: Dict[str, Any]):
         """Handle EndOfTurn events from Deepgram Flux.
@@ -571,6 +579,7 @@ class DeepgramFluxSTTService(WebsocketSTTService):
         await self.stop_processing_metrics()
         await self.push_frame(UserStoppedSpeakingFrame(), FrameDirection.DOWNSTREAM)
         await self.push_frame(UserStoppedSpeakingFrame(), FrameDirection.UPSTREAM)
+        await self._call_event_handler("on_end_of_turn", transcript)
 
     async def _handle_eager_end_of_turn(self, transcript: str, data: Dict[str, Any]):
         """Handle EagerEndOfTurn events from Deepgram Flux.
@@ -615,6 +624,7 @@ class DeepgramFluxSTTService(WebsocketSTTService):
                 result=data,
             )
         )
+        await self._call_event_handler("on_eager_end_of_turn", transcript)
 
     async def _handle_update(self, transcript: str):
         """Handle Update events from Deepgram Flux.
@@ -638,3 +648,4 @@ class DeepgramFluxSTTService(WebsocketSTTService):
             # both the "user started speaking" event and the first transcript simultaneously,
             # making this timing measurement meaningless in this context.
             # await self.stop_ttfb_metrics()
+            await self._call_event_handler("on_update", transcript)

src/pipecat/services/llm_service.py

@@ -492,11 +492,19 @@ class LLMService(AIService):
         tool_call_id: Optional[str] = None,
         text_content: Optional[str] = None,
         video_source: Optional[str] = None,
+        timeout: Optional[float] = 10.0,
     ):
         """Request an image from a user.
 
         Pushes a UserImageRequestFrame upstream to request an image from the
-        specified user.
+        specified user. The user image can then be processed by the LLM.
+
+        Use this function from a function call if you want the LLM to process
+        the image. If you expect the image to be processed by a vision service,
+        you might want to push a UserImageRequestFrame upstream directly.
+
+        .. deprecated:: 0.0.92
+            This method is deprecated, push a `UserImageRequestFrame` instead.
 
         Args:
             user_id: The ID of the user to request an image from.
@@ -504,15 +512,19 @@ class LLMService(AIService):
             tool_call_id: Optional tool call ID associated with the request.
             text_content: Optional text content/context for the image request.
             video_source: Optional video source identifier.
+            timeout: Optional timeout for the requested image to be added to the LLM context.
+
         """
+        import warnings
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "Method `request_image_frame()` is deprecated, push a `UserImageRequestFrame` instead.",
+                DeprecationWarning,
+            )
         await self.push_frame(
-            UserImageRequestFrame(
-                user_id=user_id,
-                function_name=function_name,
-                tool_call_id=tool_call_id,
-                context=text_content,
-                video_source=video_source,
-            ),
+            UserImageRequestFrame(user_id=user_id, text=text_content),
             FrameDirection.UPSTREAM,
         )
 

src/pipecat/services/moondream/vision.py

@@ -11,15 +11,17 @@ for image analysis and description generation.
 """
 
 import asyncio
-import base64
-from io import BytesIO
 from typing import AsyncGenerator, Optional
 
 from loguru import logger
 from PIL import Image
 
-from pipecat.frames.frames import ErrorFrame, Frame, TextFrame
-from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.frames.frames import (
+    ErrorFrame,
+    Frame,
+    TextFrame,
+    UserImageRawFrame,
+)
 from pipecat.services.vision_service import VisionService
 
 try:
@@ -92,16 +94,16 @@ class MoondreamService(VisionService):
             trust_remote_code=True,
             revision=revision,
             device_map={"": device},
-            torch_dtype=dtype,
+            dtype=dtype,
         ).eval()
 
         logger.debug("Loaded Moondream model")
 
-    async def run_vision(self, context: LLMContext) -> AsyncGenerator[Frame, None]:
+    async def run_vision(self, frame: UserImageRawFrame) -> AsyncGenerator[Frame, None]:
         """Analyze an image and generate a description.
 
         Args:
-            context: The context to process, containing image data.
+            frame: The image frame to process.
 
         Yields:
             Frame: TextFrame containing the generated image description, or ErrorFrame
@@ -112,45 +114,14 @@ class MoondreamService(VisionService):
             yield ErrorFrame("Moondream model not available")
             return
 
-        image_bytes = None
-        text = None
-        try:
-            messages = context.get_messages()
-            last_message = messages[-1]
-            last_message_content = last_message.get("content")
+        logger.debug(f"Analyzing image (bytes length: {len(frame.image)})")
 
-            for item in last_message_content:
-                if isinstance(item, dict):
-                    if (
-                        "image_url" in item
-                        and isinstance(item["image_url"], dict)
-                        and item["image_url"].get("url")
-                    ):
-                        image_bytes = base64.b64decode(item["image_url"]["url"].split(",")[1])
-                    elif "text" in item and isinstance(item["text"], str):
-                        text = item["text"]
-
-        except Exception as e:
-            logger.error(f"Exception during image extraction: {e}")
-            yield ErrorFrame("Failed to extract image from context")
-            return
-
-        if not image_bytes:
-            logger.error("No image found in context")
-            yield ErrorFrame("No image found in context")
-            return
-
-        logger.debug(
-            f"Analyzing image (bytes length: {len(image_bytes) if image_bytes else 'None'})"
-        )
-
-        def get_image_description(bytes: bytes, text: Optional[str]) -> str:
-            image_buffer = BytesIO(bytes)
-            image = Image.open(image_buffer)
+        def get_image_description(image_bytes: bytes, text: Optional[str]) -> str:
+            image = Image.frombytes(frame.format, frame.size, image_bytes)
             image_embeds = self._model.encode_image(image)
             description = self._model.query(image_embeds, text)["answer"]
             return description
 
-        description = await asyncio.to_thread(get_image_description, image_bytes, text)
+        description = await asyncio.to_thread(get_image_description, frame.image, frame.text)
 
         yield TextFrame(text=description)

src/pipecat/services/soniox/stt.py

@@ -49,6 +49,33 @@ END_TOKEN = "<end>"
 FINALIZED_TOKEN = "<fin>"
 
 
+class SonioxContextGeneralItem(BaseModel):
+    """Represents a key-value pair for structured general context information."""
+
+    key: str
+    value: str
+
+
+class SonioxContextTranslationTerm(BaseModel):
+    """Represents a custom translation mapping for ambiguous or domain-specific terms."""
+
+    source: str
+    target: str
+
+
+class SonioxContextObject(BaseModel):
+    """Context object for models with context_version 2, for Soniox stt-rt-v3-preview and higher.
+
+    Learn more about context in the documentation:
+    https://soniox.com/docs/stt/concepts/context
+    """
+
+    general: Optional[List[SonioxContextGeneralItem]] = None
+    text: Optional[str] = None
+    terms: Optional[List[str]] = None
+    translation_terms: Optional[List[SonioxContextTranslationTerm]] = None
+
+
 class SonioxInputParams(BaseModel):
     """Real-time transcription settings.
 
@@ -60,9 +87,9 @@ class SonioxInputParams(BaseModel):
         audio_format: Audio format to use for transcription.
         num_channels: Number of channels to use for transcription.
         language_hints: List of language hints to use for transcription.
-        context: Customization for transcription.
-        enable_non_final_tokens: Whether to enable non-final tokens. If false, only final tokens will be returned.
-        max_non_final_tokens_duration_ms: Maximum duration of non-final tokens.
+        context: Customization for transcription. String for models with context_version 1 and ContextObject for models with context_version 2.
+        enable_speaker_diarization: Whether to enable speaker diarization. Tokens are annotated with speaker IDs.
+        enable_language_identification: Whether to enable language identification. Tokens are annotated with language IDs.
         client_reference_id: Client reference ID to use for transcription.
     """
 
@@ -72,10 +99,10 @@ class SonioxInputParams(BaseModel):
     num_channels: Optional[int] = 1
 
     language_hints: Optional[List[Language]] = None
-    context: Optional[str] = None
+    context: Optional[SonioxContextObject | str] = None
 
-    enable_non_final_tokens: Optional[bool] = True
-    max_non_final_tokens_duration_ms: Optional[int] = None
+    enable_speaker_diarization: Optional[bool] = False
+    enable_language_identification: Optional[bool] = False
 
     client_reference_id: Optional[str] = None
 
@@ -173,6 +200,10 @@ class SonioxSTTService(STTService):
         # Either one or the other is required.
         enable_endpoint_detection = not self._vad_force_turn_endpoint
 
+        context = self._params.context
+        if isinstance(context, SonioxContextObject):
+            context = context.model_dump()
+
         # Send the initial configuration message.
         config = {
             "api_key": self._api_key,
@@ -182,9 +213,9 @@ class SonioxSTTService(STTService):
             "enable_endpoint_detection": enable_endpoint_detection,
             "sample_rate": self.sample_rate,
             "language_hints": _prepare_language_hints(self._params.language_hints),
-            "context": self._params.context,
-            "enable_non_final_tokens": self._params.enable_non_final_tokens,
-            "max_non_final_tokens_duration_ms": self._params.max_non_final_tokens_duration_ms,
+            "context": context,
+            "enable_speaker_diarization": self._params.enable_speaker_diarization,
+            "enable_language_identification": self._params.enable_language_identification,
             "client_reference_id": self._params.client_reference_id,
         }
 

src/pipecat/services/speechmatics/tts.py

@@ -0,0 +1,189 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Speechmatics TTS service integration."""
+
+from typing import AsyncGenerator, Optional
+from urllib.parse import urlencode
+
+import aiohttp
+from loguru import logger
+from pydantic import BaseModel
+
+from pipecat.frames.frames import (
+    ErrorFrame,
+    Frame,
+    TTSAudioRawFrame,
+    TTSStartedFrame,
+    TTSStoppedFrame,
+)
+from pipecat.services.tts_service import TTSService
+from pipecat.utils.tracing.service_decorators import traced_tts
+
+try:
+    from speechmatics.rt import __version__
+except ModuleNotFoundError as e:
+    logger.error(f"Exception: {e}")
+    logger.error(
+        "In order to use Speechmatics, you need to `pip install pipecat-ai[speechmatics]`."
+    )
+    raise Exception(f"Missing module: {e}")
+
+
+class SpeechmaticsTTSService(TTSService):
+    """Speechmatics TTS service implementation.
+
+    This service provides text-to-speech synthesis using the Speechmatics HTTP API.
+    It converts text to speech and returns raw PCM audio data for real-time playback.
+    """
+
+    SPEECHMATICS_SAMPLE_RATE = 16000
+
+    class InputParams(BaseModel):
+        """Optional input parameters for Speechmatics TTS configuration."""
+
+        pass
+
+    def __init__(
+        self,
+        *,
+        api_key: str,
+        base_url: str = "https://preview.tts.speechmatics.com",
+        voice_id: str = "sarah",
+        aiohttp_session: aiohttp.ClientSession,
+        sample_rate: Optional[int] = SPEECHMATICS_SAMPLE_RATE,
+        params: Optional[InputParams] = None,
+        **kwargs,
+    ):
+        """Initialize the Speechmatics TTS service.
+
+        Args:
+            api_key: Speechmatics API key for authentication.
+            base_url: Base URL for Speechmatics TTS API.
+            voice_id: Voice model to use for synthesis.
+            aiohttp_session: Shared aiohttp session for HTTP requests.
+            sample_rate: Audio sample rate in Hz.
+            params: Optional[InputParams]: Input parameters for the service.
+            **kwargs: Additional arguments passed to TTSService.
+        """
+        if sample_rate and sample_rate != self.SPEECHMATICS_SAMPLE_RATE:
+            logger.warning(
+                f"Speechmatics TTS only supports {self.SPEECHMATICS_SAMPLE_RATE}Hz sample rate. "
+                f"Current rate of {sample_rate}Hz may cause issues."
+            )
+        super().__init__(sample_rate=sample_rate, **kwargs)
+
+        # Service parameters
+        self._api_key: str = api_key
+        self._base_url: str = base_url
+        self._session = aiohttp_session
+
+        # Check we have required attributes
+        if not self._api_key:
+            raise ValueError("Missing Speechmatics API key")
+
+        # Default parameters
+        self._params = params or SpeechmaticsTTSService.InputParams()
+
+        # Set voice from constructor parameter
+        self.set_voice(voice_id)
+
+    def can_generate_metrics(self) -> bool:
+        """Check if this service can generate processing metrics.
+
+        Returns:
+            True, as Speechmatics service supports metrics generation.
+        """
+        return True
+
+    @traced_tts
+    async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
+        """Generate speech from text using Speechmatics' HTTP API.
+
+        Args:
+            text: The text to synthesize into speech.
+
+        Yields:
+            Frame: Audio frames containing the synthesized speech.
+        """
+        logger.debug(f"{self}: Generating TTS [{text}]")
+
+        headers = {
+            "Authorization": f"Bearer {self._api_key}",
+            "Content-Type": "application/json",
+        }
+
+        payload = {
+            "text": text,
+        }
+
+        url = _get_endpoint_url(self._base_url, self._voice_id, self.sample_rate)
+
+        try:
+            await self.start_ttfb_metrics()
+
+            async with self._session.post(url, json=payload, headers=headers) as response:
+                if response.status != 200:
+                    error_message = f"Speechmatics TTS error: HTTP {response.status}"
+                    logger.error(error_message)
+                    yield ErrorFrame(error=error_message)
+                    return
+
+                await self.start_tts_usage_metrics(text)
+
+                yield TTSStartedFrame()
+
+                # Process the response in streaming chunks
+                first_chunk = True
+                buffer = b""
+
+                async for chunk in response.content.iter_any():
+                    if not chunk:
+                        continue
+                    if first_chunk:
+                        await self.stop_ttfb_metrics()
+                        first_chunk = False
+
+                    buffer += chunk
+
+                    # Emit all complete 2-byte int16 samples from buffer
+                    if len(buffer) >= 2:
+                        complete_samples = len(buffer) // 2
+                        complete_bytes = complete_samples * 2
+
+                        audio_data = buffer[:complete_bytes]
+                        buffer = buffer[complete_bytes:]  # Keep remaining bytes for next iteration
+
+                        yield TTSAudioRawFrame(
+                            audio=audio_data,
+                            sample_rate=self.sample_rate,
+                            num_channels=1,
+                        )
+
+        except Exception as e:
+            logger.exception(f"Error generating TTS: {e}")
+            yield ErrorFrame(error=f"Speechmatics TTS error: {str(e)}")
+        finally:
+            yield TTSStoppedFrame()
+
+
+def _get_endpoint_url(base_url: str, voice: str, sample_rate: int) -> str:
+    """Format the TTS endpoint URL with voice, output format, and version params.
+
+    Args:
+        base_url: The base URL for the TTS endpoint.
+        voice: The voice model to use.
+        sample_rate: The audio sample rate.
+
+    Returns:
+        str: The formatted TTS endpoint URL.
+    """
+    query_params = {}
+    query_params["output_format"] = f"pcm_{sample_rate}"
+    query_params["sm-app"] = f"pipecat/{__version__}"
+    query = urlencode(query_params)
+
+    return f"{base_url}/generate/{voice}?{query}"

src/pipecat/services/vision_service.py

@@ -14,8 +14,7 @@ visual content.
 from abc import abstractmethod
 from typing import AsyncGenerator
 
-from pipecat.frames.frames import Frame, LLMContextFrame
-from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.frames.frames import Frame, UserImageRawFrame
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.ai_service import AIService
 
@@ -38,15 +37,15 @@ class VisionService(AIService):
         self._describe_text = None
 
     @abstractmethod
-    async def run_vision(self, context: LLMContext) -> AsyncGenerator[Frame, None]:
-        """Process the latest image in the context and generate results.
+    async def run_vision(self, frame: UserImageRawFrame) -> AsyncGenerator[Frame, None]:
+        """Process the given vision image and generate results.
 
         This method must be implemented by subclasses to provide actual computer
         vision functionality such as image description, object detection, or
         visual question answering.
 
         Args:
-            context: The context to process, containing image data.
+            frame: The image frame to process.
 
         Yields:
             Frame: Frames containing the vision analysis results, typically TextFrame
@@ -57,7 +56,7 @@ class VisionService(AIService):
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         """Process frames, handling vision image frames for analysis.
 
-        Automatically processes VisionImageRawFrame objects by calling run_vision
+        Automatically processes UserImageRawFrame objects by calling run_vision
         and handles metrics tracking. Other frames are passed through unchanged.
 
         Args:
@@ -66,9 +65,9 @@ class VisionService(AIService):
         """
         await super().process_frame(frame, direction)
 
-        if isinstance(frame, LLMContextFrame):
+        if isinstance(frame, UserImageRawFrame) and frame.text:
             await self.start_processing_metrics()
-            await self.process_generator(self.run_vision(frame.context))
+            await self.process_generator(self.run_vision(frame))
             await self.stop_processing_metrics()
         else:
             await self.push_frame(frame, direction)

src/pipecat/transports/daily/transport.py

@@ -1839,10 +1839,11 @@ class DailyInputTransport(BaseInputTransport):
         if render_frame:
             frame = UserImageRawFrame(
                 user_id=participant_id,
-                request=request_frame,
                 image=video_frame.buffer,
                 size=(video_frame.width, video_frame.height),
                 format=video_frame.color_format,
+                text=request_frame.text if request_frame else None,
+                add_to_context=request_frame.add_to_context if request_frame else None,
             )
             frame.transport_source = video_source
             await self.push_video_frame(frame)

src/pipecat/transports/smallwebrtc/transport.py

@@ -15,7 +15,7 @@ import asyncio
 import fractions
 import time
 from collections import deque
-from typing import Any, Awaitable, Callable, Optional
+from typing import Any, Awaitable, Callable, List, Optional
 
 import numpy as np
 from loguru import logger
@@ -567,7 +567,7 @@ class SmallWebRTCInputTransport(BaseInputTransport):
         self._receive_audio_task = None
         self._receive_video_task = None
         self._receive_screen_video_task = None
-        self._image_requests = {}
+        self._image_requests: List[UserImageRequestFrame] = []
 
         # Whether we have seen a StartFrame already.
         self._initialized = False
@@ -657,23 +657,27 @@ class SmallWebRTCInputTransport(BaseInputTransport):
                 if video_frame:
                     await self.push_video_frame(video_frame)
 
-                    # Check if there are any pending image requests and create UserImageRawFrame
-                    if self._image_requests:
-                        for req_id, request_frame in list(self._image_requests.items()):
-                            if request_frame.video_source == video_source:
-                                # Create UserImageRawFrame using the current video frame
-                                image_frame = UserImageRawFrame(
-                                    user_id=request_frame.user_id,
-                                    request=request_frame,
-                                    image=video_frame.image,
-                                    size=video_frame.size,
-                                    format=video_frame.format,
-                                )
-                                image_frame.transport_source = video_source
-                                # Push the frame to the pipeline
-                                await self.push_video_frame(image_frame)
-                                # Remove from pending requests
-                                del self._image_requests[req_id]
+                    # Check if there are any pending image requests and create
+                    # UserImageRawFrame. Use a shallow copy so we can remove
+                    # elements.
+                    for request_frame in self._image_requests[:]:
+                        if request_frame.video_source == video_source:
+                            # Create UserImageRawFrame using the current video frame
+                            image_frame = UserImageRawFrame(
+                                user_id=request_frame.user_id,
+                                image=video_frame.image,
+                                size=video_frame.size,
+                                format=video_frame.format,
+                                text=request_frame.text if request_frame else None,
+                                add_to_context=request_frame.add_to_context
+                                if request_frame
+                                else None,
+                            )
+                            image_frame.transport_source = video_source
+                            # Push the frame to the pipeline
+                            await self.push_video_frame(image_frame)
+                            # Remove from pending requests
+                            self._image_requests.remove(request_frame)
 
         except Exception as e:
             logger.error(f"{self} exception receiving data: {e.__class__.__name__} ({e})")
@@ -701,8 +705,7 @@ class SmallWebRTCInputTransport(BaseInputTransport):
         logger.debug(f"Requesting image from participant: {frame.user_id}")
 
         # Store the request
-        request_id = f"{frame.function_name}:{frame.tool_call_id}"
-        self._image_requests[request_id] = frame
+        self._image_requests.append(frame)
 
         # Default to camera if no source specified
         if frame.video_source is None: