openai: add retry logic to get_chat_completions()

add initial log observers

CHANGELOG.md

@@ -9,6 +9,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added retry logic to `get_chat_completions()` fo all OpenAI-based LLM
+  services.
+
+- Added new log observers `LLMLogObserver` and `TranscriptionLogObserver` that
+  can be useful for debugging your pipelines.
+
 - Added `room_url` property to `DailyTransport`.
 
 - Added `addons` argument to `DeepgramSTTService`.
@@ -30,8 +36,25 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 stt = DeepgramSTTService(..., live_options=LiveOptions(model="nova-2-general"))
 ```
 
+### Removed
+
+- Remove `TransportParams.audio_out_is_live` since it was not being used at all.
+
 ### Fixed
 
+- Fixed `match_endofsentence` support for ellipses.
+
+- Fixed an issue that would cause undesired interruptions via
+  `EmulateUserStartedSpeakingFrame` when only interim transcriptions (i.e. no
+  final transcriptions) where received.
+
+- Fixed an issue where `EndTaskFrame` was not triggering
+  `on_client_disconnected` or closing the WebSocket in FastAPI.
+
+- Fixed an issue in `DeepgramSTTService` where the `sample_rate` passed to the
+  `LiveOptions` was not being used, causing the service to use the default
+  sample rate of pipeline.
+
 - Fixed a context aggregator issue that would not append the LLM text response
   to the context if a function call happened in the same LLM turn.
 

examples/foundational/18-gstreamer-filesrc.py

@@ -38,7 +38,6 @@ async def main():
             "GStreamer",
             DailyParams(
                 audio_out_enabled=True,
-                audio_out_is_live=True,
                 camera_out_enabled=True,
                 camera_out_width=1280,
                 camera_out_height=720,

examples/foundational/30-observer.py

@@ -18,12 +18,10 @@ from pipecat.frames.frames import (
     BotStartedSpeakingFrame,
     BotStoppedSpeakingFrame,
     Frame,
-    LLMFullResponseEndFrame,
-    LLMFullResponseStartFrame,
-    LLMTextFrame,
     StartInterruptionFrame,
 )
 from pipecat.observers.base_observer import BaseObserver
+from pipecat.observers.loggers.llm_log_observer import LLMLogObserver
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
@@ -73,38 +71,6 @@ class DebugObserver(BaseObserver):
             logger.info(f"🤖 BOT STOP SPEAKING: {src} {arrow} {dst} at {time_sec:.2f}s")
 
 
-class LLMLogObserver(BaseObserver):
-    """Observer to log LLM activity to the console.
-
-    Logs all frame instances of:
-    - LLMFullResponseStartFrame (only from LLM service)
-    - LLMTextFrame
-    - LLMFullResponseEndFrame (only from LLM service)
-
-    This allows you to track when the LLM starts responding, what it generates, and when it finishes.
-    Log format: [LLM EVENT]: [details] at [timestamp]s
-    """
-
-    async def on_push_frame(
-        self,
-        src: FrameProcessor,
-        dst: FrameProcessor,
-        frame: Frame,
-        direction: FrameDirection,
-        timestamp: int,
-    ):
-        time_sec = timestamp / 1_000_000_000
-
-        # Only log start/end frames from OpenAILLMService
-        if isinstance(frame, (LLMFullResponseStartFrame, LLMFullResponseEndFrame)):
-            if isinstance(src, OpenAILLMService):
-                event = "START" if isinstance(frame, LLMFullResponseStartFrame) else "END"
-                logger.info(f"🧠 LLM {event} RESPONSE at {time_sec:.2f}s")
-        # Log all LLMTextFrames
-        elif isinstance(frame, LLMTextFrame):
-            logger.info(f"🧠 LLM GENERATING: {frame.text!r} at {time_sec:.2f}s")
-
-
 async def main():
     async with aiohttp.ClientSession() as session:
         (room_url, token) = await configure(session)

pyproject.toml

@@ -33,7 +33,8 @@ dependencies = [
     "pydantic~=2.10.5",
     "pyloudnorm~=0.1.1",
     "resampy~=0.4.3",
-    "soxr~=0.5.0"
+    "soxr~=0.5.0",
+    "tenacity~=9.0.0"
 ]
 
 [project.urls]

src/pipecat/observers/loggers/llm_log_observer.py

@@ -0,0 +1,85 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+from loguru import logger
+
+from pipecat.frames.frames import (
+    Frame,
+    FunctionCallInProgressFrame,
+    FunctionCallResultFrame,
+    LLMFullResponseEndFrame,
+    LLMFullResponseStartFrame,
+    LLMMessagesFrame,
+    LLMTextFrame,
+)
+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
+
+
+class LLMLogObserver(BaseObserver):
+    """Observer to log LLM activity to the console.
+
+    Logs all frame instances (only from/to LLM service) of:
+
+    - LLMFullResponseStartFrame
+    - LLMFullResponseEndFrame
+    - LLMTextFrame
+    - FunctionCallInProgressFrame
+    - LLMMessagesFrame
+    - OpenAILLMContextFrame
+
+    This allows you to track when the LLM starts responding, what it generates,
+    and when it finishes.
+
+    """
+
+    async def on_push_frame(
+        self,
+        src: FrameProcessor,
+        dst: FrameProcessor,
+        frame: Frame,
+        direction: FrameDirection,
+        timestamp: int,
+    ):
+        if not isinstance(src, LLMService) and not isinstance(dst, LLMService):
+            return
+
+        time_sec = timestamp / 1_000_000_000
+
+        arrow = "→"
+
+        # Log LLM start/end frames (output)
+        if isinstance(frame, (LLMFullResponseStartFrame, LLMFullResponseEndFrame)):
+            event = "START" if isinstance(frame, LLMFullResponseStartFrame) else "END"
+            logger.debug(f"🧠 {src} {arrow} LLM {event} RESPONSE at {time_sec:.2f}s")
+        # Log all LLMTextFrames (output)
+        elif isinstance(frame, LLMTextFrame):
+            logger.debug(f"🧠 {src} {arrow} LLM GENERATING: {frame.text!r} at {time_sec:.2f}s")
+        # Log function calling (output)
+        elif (
+            isinstance(frame, FunctionCallInProgressFrame)
+            and direction != FrameDirection.DOWNSTREAM
+        ):
+            logger.debug(
+                f"🧠 {src} {arrow} LLM FUNCTION CALL ({frame.tool_call_id}): {frame.function_name!r}({frame.arguments}) at {time_sec:.2f}s"
+            )
+        # Log LLMMessagesFrame (input)
+        elif isinstance(frame, LLMMessagesFrame):
+            logger.debug(
+                f"🧠 {arrow} {dst} LLM MESSAGES FRAME: {frame.messages} at {time_sec:.2f}s"
+            )
+        # Log OpenAILLMContextFrame (input)
+        elif isinstance(frame, OpenAILLMContextFrame):
+            logger.debug(
+                f"🧠 {arrow} {dst} LLM CONTEXT FRAME: {frame.context.messages} at {time_sec:.2f}s"
+            )
+        # Log function call result (input)
+        elif isinstance(frame, FunctionCallResultFrame):
+            logger.debug(
+                f"🧠 {arrow} {src} LLM FUNCTION CALL RESULT ({frame.tool_call_id}): {frame.result} at {time_sec:.2f}s"
+            )

src/pipecat/observers/loggers/transcription_log_observer.py

@@ -0,0 +1,54 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+from loguru import logger
+
+from pipecat.frames.frames import (
+    Frame,
+    InterimTranscriptionFrame,
+    TranscriptionFrame,
+)
+from pipecat.observers.base_observer import BaseObserver
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.services.ai_services import STTService
+
+
+class TranscriptionLogObserver(BaseObserver):
+    """Observer to log transcription activity to the console.
+
+    Logs all frame instances (only from STT service) of:
+
+    - TranscriptionFrame
+    - InterimTranscriptionFrame
+
+    This allows you to track when the LLM starts responding, what it generates,
+    and when it finishes.
+
+    """
+
+    async def on_push_frame(
+        self,
+        src: FrameProcessor,
+        dst: FrameProcessor,
+        frame: Frame,
+        direction: FrameDirection,
+        timestamp: int,
+    ):
+        if not isinstance(src, STTService):
+            return
+
+        time_sec = timestamp / 1_000_000_000
+
+        arrow = "→"
+
+        if isinstance(frame, TranscriptionFrame):
+            logger.debug(
+                f"💬 {src} {arrow} TRANSCRIPTION: {frame.text!r} from {frame.user_id!r} at {time_sec:.2f}s"
+            )
+        elif isinstance(frame, InterimTranscriptionFrame):
+            logger.debug(
+                f"💬 {src} {arrow} INTERIM TRANSCRIPTION: {frame.text!r} from {frame.user_id!r} at {time_sec:.2f}s"
+            )

src/pipecat/processors/aggregators/llm_response.py

@@ -10,13 +10,14 @@ from abc import abstractmethod
 from typing import List
 
 from pipecat.frames.frames import (
-    BotStoppedSpeakingFrame,
     CancelFrame,
     EmulateUserStartedSpeakingFrame,
     EmulateUserStoppedSpeakingFrame,
     EndFrame,
     Frame,
     InterimTranscriptionFrame,
+    LLMFullResponseEndFrame,
+    LLMFullResponseStartFrame,
     LLMMessagesAppendFrame,
     LLMMessagesFrame,
     LLMMessagesUpdateFrame,
@@ -25,7 +26,6 @@ from pipecat.frames.frames import (
     StartInterruptionFrame,
     TextFrame,
     TranscriptionFrame,
-    TTSTextFrame,
     UserStartedSpeakingFrame,
     UserStoppedSpeakingFrame,
 )
@@ -293,7 +293,13 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
             await self.push_aggregation()
 
     async def _handle_transcription(self, frame: TranscriptionFrame):
-        self._aggregation += f" {frame.text}" if self._aggregation else frame.text
+        text = frame.text
+
+        # Make sure we really have some text.
+        if not text.strip():
+            return
+
+        self._aggregation += f" {text}" if self._aggregation else text
         # We just got a final result, so let's reset interim results.
         self._seen_interim_results = False
         # Reset aggregation timer.
@@ -301,8 +307,6 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
 
     async def _handle_interim_transcription(self, _: InterimTranscriptionFrame):
         self._seen_interim_results = True
-        # Reset aggregation timer.
-        self._aggregation_event.set()
 
     def _create_aggregation_task(self):
         self._aggregation_task = self.create_task(self._aggregation_task_handler())
@@ -352,8 +356,8 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
 
 class LLMAssistantContextAggregator(LLMContextResponseAggregator):
     """This is an assistant LLM aggregator that uses an LLM context to store the
-    conversation. It aggregates text frames spoken by the TTS service and pushes
-    the context when the bot stops speaking..
+    conversation. It aggregates text frames received between
+    `LLMFullResponseStartFrame` and `LLMFullResponseEndFrame`.
 
     """
 
@@ -361,6 +365,8 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
         super().__init__(context=context, role="assistant", **kwargs)
         self._expect_stripped_words = expect_stripped_words
 
+        self._started = False
+
         self.reset()
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
@@ -371,10 +377,11 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
             # Reset anyways
             self.reset()
             await self.push_frame(frame, direction)
-        elif isinstance(frame, BotStoppedSpeakingFrame):
-            await self._handle_bot_stopped_speaking(frame)
-            await self.push_frame(frame, direction)
-        elif isinstance(frame, TTSTextFrame):
+        elif isinstance(frame, LLMFullResponseStartFrame):
+            await self._handle_llm_start(frame)
+        elif isinstance(frame, LLMFullResponseEndFrame):
+            await self._handle_llm_end(frame)
+        elif isinstance(frame, TextFrame):
             await self._handle_text(frame)
         elif isinstance(frame, LLMMessagesAppendFrame):
             self.add_messages(frame.messages)
@@ -385,10 +392,17 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
         else:
             await self.push_frame(frame, direction)
 
-    async def _handle_bot_stopped_speaking(self, _: BotStoppedSpeakingFrame):
+    async def _handle_llm_start(self, _: LLMFullResponseStartFrame):
+        self._started = True
+
+    async def _handle_llm_end(self, _: LLMFullResponseEndFrame):
+        self._started = False
         await self.push_aggregation()
 
     async def _handle_text(self, frame: TextFrame):
+        if not self._started:
+            return
+
         if self._expect_stripped_words:
             self._aggregation += f" {frame.text}" if self._aggregation else frame.text
         else:

src/pipecat/services/cerebras.py

@@ -7,6 +7,7 @@
 from typing import List
 
 from loguru import logger
+from tenacity import retry, stop_after_attempt, wait_fixed
 
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.services.openai import OpenAILLMService
@@ -52,6 +53,7 @@ class CerebrasLLMService(OpenAILLMService):
         logger.debug(f"Creating Cerebras client with api {base_url}")
         return super().create_client(api_key, base_url, **kwargs)
 
+    @retry(stop=stop_after_attempt(2), wait=wait_fixed(2))
     async def get_chat_completions(
         self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
     ) -> AsyncStream[ChatCompletionChunk]:

src/pipecat/services/deepgram.py

@@ -124,6 +124,7 @@ class DeepgramSTTService(STTService):
         addons: Optional[Dict] = None,
         **kwargs,
     ):
+        sample_rate = sample_rate or (live_options.sample_rate if live_options else None)
         super().__init__(sample_rate=sample_rate, **kwargs)
 
         default_options = LiveOptions(

src/pipecat/services/deepseek.py

@@ -8,6 +8,7 @@
 from typing import List
 
 from loguru import logger
+from tenacity import retry, stop_after_attempt, wait_fixed
 
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.services.openai import OpenAILLMService
@@ -53,6 +54,7 @@ class DeepSeekLLMService(OpenAILLMService):
         logger.debug(f"Creating DeepSeek client with api {base_url}")
         return super().create_client(api_key, base_url, **kwargs)
 
+    @retry(stop=stop_after_attempt(2), wait=wait_fixed(2))
     async def get_chat_completions(
         self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
     ) -> AsyncStream[ChatCompletionChunk]:

src/pipecat/services/fireworks.py

@@ -8,6 +8,7 @@
 from typing import List
 
 from loguru import logger
+from tenacity import retry, stop_after_attempt, wait_fixed
 
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.services.openai import OpenAILLMService
@@ -50,6 +51,7 @@ class FireworksLLMService(OpenAILLMService):
         logger.debug(f"Creating Fireworks client with api {base_url}")
         return super().create_client(api_key, base_url, **kwargs)
 
+    @retry(stop=stop_after_attempt(2), wait=wait_fixed(2))
     async def get_chat_completions(
         self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
     ):

src/pipecat/services/openai.py

@@ -15,6 +15,7 @@ import httpx
 from loguru import logger
 from PIL import Image
 from pydantic import BaseModel, Field
+from tenacity import retry, stop_after_attempt, wait_fixed
 
 from pipecat.frames.frames import (
     ErrorFrame,
@@ -160,6 +161,7 @@ class BaseOpenAILLMService(LLMService):
     def can_generate_metrics(self) -> bool:
         return True
 
+    @retry(stop=stop_after_attempt(2), wait=wait_fixed(2))
     async def get_chat_completions(
         self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
     ) -> AsyncStream[ChatCompletionChunk]:

src/pipecat/services/openpipe.py

@@ -7,6 +7,7 @@
 from typing import Dict, List, Optional
 
 from loguru import logger
+from tenacity import retry, stop_after_attempt, wait_fixed
 
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.services.openai import OpenAILLMService
@@ -55,6 +56,7 @@ class OpenPipeLLMService(OpenAILLMService):
         )
         return client
 
+    @retry(stop=stop_after_attempt(2), wait=wait_fixed(2))
     async def get_chat_completions(
         self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
     ) -> AsyncStream[ChatCompletionChunk]:

src/pipecat/services/perplexity.py

@@ -7,6 +7,7 @@
 from typing import List
 
 from loguru import logger
+from tenacity import retry, stop_after_attempt, wait_fixed
 
 from pipecat.metrics.metrics import LLMTokenUsage
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
@@ -56,6 +57,7 @@ class PerplexityLLMService(OpenAILLMService):
         self._has_reported_prompt_tokens = False
         self._is_processing = False
 
+    @retry(stop=stop_after_attempt(2), wait=wait_fixed(2))
     async def get_chat_completions(
         self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
     ) -> AsyncStream[ChatCompletionChunk]:

src/pipecat/transports/base_transport.py

@@ -4,7 +4,6 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-import asyncio
 import inspect
 from abc import ABC, abstractmethod
 from typing import Optional
@@ -30,7 +29,6 @@ class TransportParams(BaseModel):
     camera_out_framerate: int = 30
     camera_out_color_format: str = "RGB"
     audio_out_enabled: bool = False
-    audio_out_is_live: bool = False
     audio_out_sample_rate: Optional[int] = None
     audio_out_channels: int = 1
     audio_out_bitrate: int = 96000

src/pipecat/transports/network/fastapi_websocket.py

@@ -55,45 +55,89 @@ class FastAPIWebsocketCallbacks(BaseModel):
     on_session_timeout: Callable[[WebSocket], Awaitable[None]]
 
 
+class FastAPIWebsocketClient:
+    def __init__(self, websocket: WebSocket, is_binary: bool, callbacks: FastAPIWebsocketCallbacks):
+        self._websocket = websocket
+        self._closing = False
+        self._is_binary = is_binary
+        self._callbacks = callbacks
+
+    def receive(self) -> typing.AsyncIterator[bytes | str]:
+        return self._websocket.iter_bytes() if self._is_binary else self._websocket.iter_text()
+
+    async def send(self, data: str | bytes):
+        if self._can_send():
+            if self._is_binary:
+                await self._websocket.send_bytes(data)
+            else:
+                await self._websocket.send_text(data)
+
+    async def disconnect(self):
+        if self.is_connected and not self.is_closing:
+            self._closing = True
+            await self._websocket.close()
+            await self.trigger_client_disconnected()
+
+    async def trigger_client_disconnected(self):
+        await self._callbacks.on_client_disconnected(self._websocket)
+
+    async def trigger_client_connected(self):
+        await self._callbacks.on_client_connected(self._websocket)
+
+    async def trigger_client_timout(self):
+        await self._callbacks.on_session_timeout(self._websocket)
+
+    def _can_send(self):
+        return self.is_connected and not self.is_closing
+
+    @property
+    def is_connected(self) -> bool:
+        return self._websocket.client_state == WebSocketState.CONNECTED
+
+    @property
+    def is_closing(self) -> bool:
+        return self._closing
+
+
 class FastAPIWebsocketInputTransport(BaseInputTransport):
     def __init__(
         self,
-        websocket: WebSocket,
+        client: FastAPIWebsocketClient,
         params: FastAPIWebsocketParams,
-        callbacks: FastAPIWebsocketCallbacks,
         **kwargs,
     ):
         super().__init__(params, **kwargs)
-
-        self._websocket = websocket
+        self._client = client
         self._params = params
-        self._callbacks = callbacks
+        self._receive_task = None
+        self._monitor_websocket_task = None
 
     async def start(self, frame: StartFrame):
         await super().start(frame)
         await self._params.serializer.setup(frame)
         if self._params.session_timeout:
             self._monitor_websocket_task = self.create_task(self._monitor_websocket())
-        await self._callbacks.on_client_connected(self._websocket)
+        await self._client.trigger_client_connected()
         self._receive_task = self.create_task(self._receive_messages())
 
+    async def _stop_tasks(self):
+        if self._monitor_websocket_task:
+            await self.cancel_task(self._monitor_websocket_task)
+        await self.cancel_task(self._receive_task)
+
     async def stop(self, frame: EndFrame):
         await super().stop(frame)
-        await self.cancel_task(self._receive_task)
+        await self._stop_tasks()
+        await self._client.disconnect()
 
     async def cancel(self, frame: CancelFrame):
         await super().cancel(frame)
-        await self.cancel_task(self._receive_task)
-
-    def _iter_data(self) -> typing.AsyncIterator[bytes | str]:
-        if self._params.serializer.type == FrameSerializerType.BINARY:
-            return self._websocket.iter_bytes()
-        else:
-            return self._websocket.iter_text()
+        await self._stop_tasks()
+        await self._client.disconnect()
 
     async def _receive_messages(self):
         try:
-            async for message in self._iter_data():
+            async for message in self._client.receive():
                 frame = await self._params.serializer.deserialize(message)
 
                 if not frame:
@@ -106,19 +150,23 @@ class FastAPIWebsocketInputTransport(BaseInputTransport):
         except Exception as e:
             logger.error(f"{self} exception receiving data: {e.__class__.__name__} ({e})")
 
-        await self._callbacks.on_client_disconnected(self._websocket)
+        await self._client.trigger_client_disconnected()
 
     async def _monitor_websocket(self):
         """Wait for self._params.session_timeout seconds, if the websocket is still open, trigger timeout event."""
         await asyncio.sleep(self._params.session_timeout)
-        await self._callbacks.on_session_timeout(self._websocket)
+        await self._client.trigger_client_timout()
 
 
 class FastAPIWebsocketOutputTransport(BaseOutputTransport):
-    def __init__(self, websocket: WebSocket, params: FastAPIWebsocketParams, **kwargs):
+    def __init__(
+        self,
+        client: FastAPIWebsocketClient,
+        params: FastAPIWebsocketParams,
+        **kwargs,
+    ):
         super().__init__(params, **kwargs)
-
-        self._websocket = websocket
+        self._client = client
         self._params = params
 
         # write_raw_audio_frames() is called quickly, as soon as we get audio
@@ -134,6 +182,14 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
         await self._params.serializer.setup(frame)
         self._send_interval = (self._audio_chunk_size / self.sample_rate) / 2
 
+    async def stop(self, frame: EndFrame):
+        await super().stop(frame)
+        await self._client.disconnect()
+
+    async def cancel(self, frame: CancelFrame):
+        await super().cancel(frame)
+        await self._client.disconnect()
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
 
@@ -145,7 +201,10 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
         await self._write_frame(frame)
 
     async def write_raw_audio_frames(self, frames: bytes):
-        if self._websocket.client_state != WebSocketState.CONNECTED:
+        if self._client.is_closing:
+            return
+
+        if not self._client.is_connected:
             # Simulate audio playback with a sleep.
             await self._write_audio_sleep()
             return
@@ -172,25 +231,17 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
 
         await self._write_frame(frame)
 
-        self._websocket_audio_buffer = bytes()
-
         # Simulate audio playback with a sleep.
         await self._write_audio_sleep()
 
     async def _write_frame(self, frame: Frame):
         try:
             payload = await self._params.serializer.serialize(frame)
-            if payload and self._websocket.client_state == WebSocketState.CONNECTED:
-                await self._send_data(payload)
+            if payload:
+                await self._client.send(payload)
         except Exception as e:
             logger.error(f"{self} exception sending data: {e.__class__.__name__} ({e})")
 
-    def _send_data(self, data: str | bytes):
-        if self._params.serializer.type == FrameSerializerType.BINARY:
-            return self._websocket.send_bytes(data)
-        else:
-            return self._websocket.send_text(data)
-
     async def _write_audio_sleep(self):
         # Simulate a clock.
         current_time = time.monotonic()
@@ -219,11 +270,14 @@ class FastAPIWebsocketTransport(BaseTransport):
             on_session_timeout=self._on_session_timeout,
         )
 
+        is_binary = self._params.serializer.type == FrameSerializerType.BINARY
+        self._client = FastAPIWebsocketClient(websocket, is_binary, self._callbacks)
+
         self._input = FastAPIWebsocketInputTransport(
-            websocket, self._params, self._callbacks, name=self._input_name
+            self._client, self._params, name=self._input_name
         )
         self._output = FastAPIWebsocketOutputTransport(
-            websocket, self._params, name=self._output_name
+            self._client, self._params, name=self._output_name
         )
 
         # Register supported handlers. The user will only be able to register

src/pipecat/utils/string.py

@@ -13,8 +13,8 @@ ENDOFSENTENCE_PATTERN_STR = r"""
     (?<!Mr|Ms|Dr)    # Negative lookbehind: not preceded by Mr, Ms, Dr (combined bc. length is the same)
     (?<!Mrs)         # Negative lookbehind: not preceded by "Mrs"
     (?<!Prof)        # Negative lookbehind: not preceded by "Prof"
-    [\.\?\!;]|       # Match a period, question mark, exclamation point, or semicolon
-    [。？！；।]        # the full-width version (mainly used in East Asian languages such as Chinese, Hindi)
+    (\.\s*\.\s*\.|[\.\?\!;])|   # Match a period, question mark, exclamation point, or semicolon
+    (\。\s*\。\s*\。|[。？！；।])  # the full-width version (mainly used in East Asian languages such as Chinese, Hindi)
     $                # End of string
 """
 ENDOFSENTENCE_PATTERN = re.compile(ENDOFSENTENCE_PATTERN_STR, re.VERBOSE)

tests/test_context_aggregators.py

@@ -9,14 +9,15 @@ import unittest
 import google.ai.generativelanguage as glm
 
 from pipecat.frames.frames import (
-    BotStoppedSpeakingFrame,
     EmulateUserStartedSpeakingFrame,
     EmulateUserStoppedSpeakingFrame,
     InterimTranscriptionFrame,
+    LLMFullResponseEndFrame,
+    LLMFullResponseStartFrame,
     OpenAILLMContextAssistantTimestampFrame,
     StartInterruptionFrame,
+    TextFrame,
     TranscriptionFrame,
-    TTSTextFrame,
     UserStartedSpeakingFrame,
     UserStoppedSpeakingFrame,
 )
@@ -427,6 +428,20 @@ class BaseTestAssistantContextAggreagator:
     ):
         assert context.messages[index]["content"] == content
 
+    async def test_empty(self):
+        assert self.CONTEXT_CLASS is not None, "CONTEXT_CLASS must be set in a subclass"
+        assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass"
+
+        context = self.CONTEXT_CLASS()
+        aggregator = self.AGGREGATOR_CLASS(context)
+        frames_to_send = [LLMFullResponseStartFrame(), LLMFullResponseEndFrame()]
+        expected_down_frames = []
+        await run_test(
+            aggregator,
+            frames_to_send=frames_to_send,
+            expected_down_frames=expected_down_frames,
+        )
+
     async def test_single_text(self):
         assert self.CONTEXT_CLASS is not None, "CONTEXT_CLASS must be set in a subclass"
         assert self.AGGREGATOR_CLASS is not None, "AGGREGATOR_CLASS must be set in a subclass"
@@ -434,11 +449,11 @@ class BaseTestAssistantContextAggreagator:
         context = self.CONTEXT_CLASS()
         aggregator = self.AGGREGATOR_CLASS(context)
         frames_to_send = [
-            TTSTextFrame(text="Hello Pipecat!"),
-            SleepFrame(),
-            BotStoppedSpeakingFrame(),
+            LLMFullResponseStartFrame(),
+            TextFrame(text="Hello Pipecat!"),
+            LLMFullResponseEndFrame(),
         ]
-        expected_down_frames = [BotStoppedSpeakingFrame, *self.EXPECTED_CONTEXT_FRAMES]
+        expected_down_frames = [*self.EXPECTED_CONTEXT_FRAMES]
         await run_test(
             aggregator,
             frames_to_send=frames_to_send,
@@ -453,14 +468,14 @@ class BaseTestAssistantContextAggreagator:
         context = self.CONTEXT_CLASS()
         aggregator = self.AGGREGATOR_CLASS(context, expect_stripped_words=False)
         frames_to_send = [
-            TTSTextFrame(text="Hello "),
-            TTSTextFrame(text="Pipecat. "),
-            TTSTextFrame(text="How are "),
-            TTSTextFrame(text="you?"),
-            SleepFrame(),
-            BotStoppedSpeakingFrame(),
+            LLMFullResponseStartFrame(),
+            TextFrame(text="Hello "),
+            TextFrame(text="Pipecat. "),
+            TextFrame(text="How are "),
+            TextFrame(text="you?"),
+            LLMFullResponseEndFrame(),
         ]
-        expected_down_frames = [BotStoppedSpeakingFrame, *self.EXPECTED_CONTEXT_FRAMES]
+        expected_down_frames = [*self.EXPECTED_CONTEXT_FRAMES]
         await run_test(
             aggregator,
             frames_to_send=frames_to_send,
@@ -475,14 +490,14 @@ class BaseTestAssistantContextAggreagator:
         context = self.CONTEXT_CLASS()
         aggregator = self.AGGREGATOR_CLASS(context)
         frames_to_send = [
-            TTSTextFrame(text="Hello"),
-            TTSTextFrame(text="Pipecat."),
-            TTSTextFrame(text="How are"),
-            TTSTextFrame(text="you?"),
-            SleepFrame(),
-            BotStoppedSpeakingFrame(),
+            LLMFullResponseStartFrame(),
+            TextFrame(text="Hello"),
+            TextFrame(text="Pipecat."),
+            TextFrame(text="How are"),
+            TextFrame(text="you?"),
+            LLMFullResponseEndFrame(),
         ]
-        expected_down_frames = [BotStoppedSpeakingFrame, *self.EXPECTED_CONTEXT_FRAMES]
+        expected_down_frames = [*self.EXPECTED_CONTEXT_FRAMES]
         await run_test(
             aggregator,
             frames_to_send=frames_to_send,
@@ -497,21 +512,16 @@ class BaseTestAssistantContextAggreagator:
         context = self.CONTEXT_CLASS()
         aggregator = self.AGGREGATOR_CLASS(context, expect_stripped_words=False)
         frames_to_send = [
-            TTSTextFrame(text="Hello "),
-            TTSTextFrame(text="Pipecat."),
-            SleepFrame(),
-            BotStoppedSpeakingFrame(),
-            TTSTextFrame(text="How are "),
-            TTSTextFrame(text="you?"),
-            SleepFrame(),
-            BotStoppedSpeakingFrame(),
-        ]
-        expected_down_frames = [
-            BotStoppedSpeakingFrame,
-            *self.EXPECTED_CONTEXT_FRAMES,
-            BotStoppedSpeakingFrame,
-            *self.EXPECTED_CONTEXT_FRAMES,
+            LLMFullResponseStartFrame(),
+            TextFrame(text="Hello "),
+            TextFrame(text="Pipecat."),
+            LLMFullResponseEndFrame(),
+            LLMFullResponseStartFrame(),
+            TextFrame(text="How are "),
+            TextFrame(text="you?"),
+            LLMFullResponseEndFrame(),
         ]
+        expected_down_frames = [*self.EXPECTED_CONTEXT_FRAMES, *self.EXPECTED_CONTEXT_FRAMES]
         await run_test(
             aggregator,
             frames_to_send=frames_to_send,
@@ -527,22 +537,20 @@ class BaseTestAssistantContextAggreagator:
         context = self.CONTEXT_CLASS()
         aggregator = self.AGGREGATOR_CLASS(context, expect_stripped_words=False)
         frames_to_send = [
-            TTSTextFrame(text="Hello "),
-            TTSTextFrame(text="Pipecat."),
-            SleepFrame(),
-            BotStoppedSpeakingFrame(),
+            LLMFullResponseStartFrame(),
+            TextFrame(text="Hello "),
+            TextFrame(text="Pipecat."),
+            LLMFullResponseEndFrame(),
             SleepFrame(AGGREGATION_SLEEP),
             StartInterruptionFrame(),
-            TTSTextFrame(text="How are "),
-            TTSTextFrame(text="you?"),
-            SleepFrame(),
-            BotStoppedSpeakingFrame(),
+            LLMFullResponseStartFrame(),
+            TextFrame(text="How are "),
+            TextFrame(text="you?"),
+            LLMFullResponseEndFrame(),
         ]
         expected_down_frames = [
-            BotStoppedSpeakingFrame,
             *self.EXPECTED_CONTEXT_FRAMES,
             StartInterruptionFrame,
-            BotStoppedSpeakingFrame,
             *self.EXPECTED_CONTEXT_FRAMES,
         ]
         await run_test(

tests/test_utils_string.py

@@ -11,10 +11,12 @@ from pipecat.utils.string import match_endofsentence
 
 class TestUtilsString(unittest.IsolatedAsyncioTestCase):
     async def test_endofsentence(self):
-        assert match_endofsentence("This is a sentence.")
-        assert match_endofsentence("This is a sentence! ")
-        assert match_endofsentence("This is a sentence?")
-        assert match_endofsentence("This is a sentence;")
+        assert match_endofsentence("This is a sentence.") == 19
+        assert match_endofsentence("This is a sentence!") == 19
+        assert match_endofsentence("This is a sentence?") == 19
+        assert match_endofsentence("This is a sentence;") == 19
+        assert match_endofsentence("This is a sentence...") == 21
+        assert match_endofsentence("This is a sentence . . .") == 24
         assert not match_endofsentence("This is not a sentence")
         assert not match_endofsentence("This is not a sentence,")
         assert not match_endofsentence("This is not a sentence, ")