Merge pull request #996 from pipecat-ai/aleix/introduce-observers

introduce pipeline frame observers

CHANGELOG.md

@@ -9,9 +9,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Introduced pipeline frame observers. Observers can view all the frames that go
+  through the pipeline without the need to inject processors in the
+  pipeline. This can be useful, for example, to implement frame loggers or
+  debuggers among other things.
+
 - Added `OpenRouter` for OpenRouter integration with an OpenAI-compatible
   interface. Added foundational example `14m-function-calling-openrouter.py`.
-  
+
 - Added a new `WebsocketService` based class for TTS services, containing
   base functions and retry logic.
 

examples/simple-chatbot/server/bot-gemini.py

@@ -41,17 +41,8 @@ from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
-from pipecat.processors.frameworks.rtvi import (
+from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIProcessor
-    RTVIBotTranscriptionProcessor,
-    RTVIConfig,
-    RTVIMetricsProcessor,
-    RTVIProcessor,
-    RTVISpeakingProcessor,
-    RTVIUserTranscriptionProcessor,
-)
-from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.gemini_multimodal_live.gemini import GeminiMultimodalLiveLLMService
-from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 
 load_dotenv(override=True)
@@ -168,20 +159,6 @@ async def main():
         #
         # RTVI events for Pipecat client UI
         #
-
-        # This will send `user-*-speaking` and `bot-*-speaking` messages.
-        rtvi_speaking = RTVISpeakingProcessor()
-
-        # This will emit UserTranscript events.
-        rtvi_user_transcription = RTVIUserTranscriptionProcessor()
-
-        # This will emit BotTranscript events.
-        rtvi_bot_transcription = RTVIBotTranscriptionProcessor()
-
-        # This will send `metrics` messages.
-        rtvi_metrics = RTVIMetricsProcessor()
-
-        # Handles RTVI messages from the client
         rtvi = RTVIProcessor(config=RTVIConfig(config=[]))
 
         pipeline = Pipeline(
@@ -190,11 +167,7 @@ async def main():
                 rtvi,
                 context_aggregator.user(),
                 llm,
-                rtvi_speaking,
-                rtvi_user_transcription,
-                rtvi_bot_transcription,
                 ta,
-                rtvi_metrics,
                 transport.output(),
                 context_aggregator.assistant(),
             ]
@@ -206,6 +179,7 @@ async def main():
                 allow_interruptions=True,
                 enable_metrics=True,
                 enable_usage_metrics=True,
+                observers=[rtvi.observer()],
             ),
         )
         await task.queue_frame(quiet_frame)

examples/simple-chatbot/server/bot-openai.py

@@ -41,14 +41,7 @@ from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
-from pipecat.processors.frameworks.rtvi import (
+from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIProcessor
-    RTVIBotTranscriptionProcessor,
-    RTVIConfig,
-    RTVIMetricsProcessor,
-    RTVIProcessor,
-    RTVISpeakingProcessor,
-    RTVIUserTranscriptionProcessor,
-)
 from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
@@ -189,34 +182,16 @@ async def main():
         #
         # RTVI events for Pipecat client UI
         #
-
-        # This will send `user-*-speaking` and `bot-*-speaking` messages.
-        rtvi_speaking = RTVISpeakingProcessor()
-
-        # This will emit UserTranscript events.
-        rtvi_user_transcription = RTVIUserTranscriptionProcessor()
-
-        # This will emit BotTranscript events.
-        rtvi_bot_transcription = RTVIBotTranscriptionProcessor()
-
-        # This will send `metrics` messages.
-        rtvi_metrics = RTVIMetricsProcessor()
-
-        # Handles RTVI messages from the client
         rtvi = RTVIProcessor(config=RTVIConfig(config=[]))
 
         pipeline = Pipeline(
             [
                 transport.input(),
                 rtvi,
-                rtvi_speaking,
-                rtvi_user_transcription,
                 context_aggregator.user(),
                 llm,
-                rtvi_bot_transcription,
                 tts,
                 ta,
-                rtvi_metrics,
                 transport.output(),
                 context_aggregator.assistant(),
             ]
@@ -228,6 +203,7 @@ async def main():
                 allow_interruptions=True,
                 enable_metrics=True,
                 enable_usage_metrics=True,
+                observers=[rtvi.observer()],
             ),
         )
         await task.queue_frame(quiet_frame)

src/pipecat/frames/frames.py

@@ -5,7 +5,7 @@
 #
 
 from dataclasses import dataclass, field
-from typing import Any, Awaitable, Callable, List, Literal, Mapping, Optional, Tuple
+from typing import TYPE_CHECKING, Any, Awaitable, Callable, List, Literal, Mapping, Optional, Tuple
 
 from pipecat.audio.vad.vad_analyzer import VADParams
 from pipecat.clocks.base_clock import BaseClock
@@ -14,6 +14,9 @@ from pipecat.transcriptions.language import Language
 from pipecat.utils.time import nanoseconds_to_str
 from pipecat.utils.utils import obj_count, obj_id
 
+if TYPE_CHECKING:
+    from pipecat.observers.base_observer import BaseObserver
+
 
 def format_pts(pts: int | None):
     return nanoseconds_to_str(pts) if pts else None
@@ -177,6 +180,20 @@ class TextFrame(DataFrame):
         return f"{self.name}(pts: {pts}, text: [{self.text}])"
 
 
+@dataclass
+class LLMTextFrame(TextFrame):
+    """A text frame generated by LLM services."""
+
+    pass
+
+
+@dataclass
+class TTSTextFrame(TextFrame):
+    """A text frame generated by TTS services."""
+
+    pass
+
+
 @dataclass
 class TranscriptionFrame(TextFrame):
     """A text frame with transcription-specific data. Will be placed in the
@@ -372,6 +389,7 @@ class StartFrame(SystemFrame):
     enable_metrics: bool = False
     enable_usage_metrics: bool = False
     report_only_initial_ttfb: bool = False
+    observer: Optional["BaseObserver"] = None
 
 
 @dataclass

src/pipecat/observers/base_observer.py

@@ -0,0 +1,44 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+from abc import ABC, abstractmethod
+
+from pipecat.frames.frames import Frame
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+
+
+class BaseObserver(ABC):
+    """This is the base class for pipeline frame observers. Observers can view
+    all the frames that go through the pipeline without the need to inject
+    processors in the pipeline. This can be useful, for example, to implement
+    frame loggers or debuggers among other things.
+
+    """
+
+    @abstractmethod
+    async def on_push_frame(
+        self,
+        src: FrameProcessor,
+        dst: FrameProcessor,
+        frame: Frame,
+        direction: FrameDirection,
+        timestamp: int,
+    ):
+        """Abstract method to handle the event when a frame is pushed from one
+        processor to another.
+
+        Args:
+            src (FrameProcessor): The source frame processor that is sending the frame.
+            dst (FrameProcessor): The destination frame processor that will receive the frame.
+            frame (Frame): The frame being transferred between processors.
+            direction (FrameDirection): The direction of the frame transfer.
+            timestamp (int): The timestamp when the frame was pushed (based on the pipeline clock).
+
+        This method should be implemented by subclasses to define specific behavior
+        when a frame is pushed.
+
+        """
+        pass

src/pipecat/pipeline/task.py

@@ -5,10 +5,10 @@
 #
 
 import asyncio
-from typing import AsyncIterable, Iterable
+from typing import AsyncIterable, Iterable, List
 
 from loguru import logger
-from pydantic import BaseModel
+from pydantic import BaseModel, ConfigDict
 
 from pipecat.clocks.base_clock import BaseClock
 from pipecat.clocks.system_clock import SystemClock
@@ -24,20 +24,31 @@ from pipecat.frames.frames import (
     StopTaskFrame,
 )
 from pipecat.metrics.metrics import ProcessingMetricsData, TTFBMetricsData
+from pipecat.observers.base_observer import BaseObserver
 from pipecat.pipeline.base_pipeline import BasePipeline
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.utils.utils import obj_count, obj_id
 
 
 class PipelineParams(BaseModel):
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
     allow_interruptions: bool = False
     enable_metrics: bool = False
     enable_usage_metrics: bool = False
     send_initial_empty_metrics: bool = True
     report_only_initial_ttfb: bool = False
+    observers: List[BaseObserver] = []
 
 
 class Source(FrameProcessor):
+    """This is the source processor that is linked at the beginning of the
+    pipeline given to the pipeline task. It allows us to easily push frames
+    downstream to the pipeline and also receive upstream frames coming from the
+    pipeline.
+
+    """
+
     def __init__(self, up_queue: asyncio.Queue):
         super().__init__()
         self._up_queue = up_queue
@@ -68,6 +79,12 @@ class Source(FrameProcessor):
 
 
 class Sink(FrameProcessor):
+    """This is the sink processor that is linked at the end of the pipeline
+    given to the pipeline task. It allows us to receive downstream frames and
+    act on them, for example, waiting to receive an EndFrame.
+
+    """
+
     def __init__(self, down_queue: asyncio.Queue):
         super().__init__()
         self._down_queue = down_queue
@@ -80,6 +97,29 @@ class Sink(FrameProcessor):
             await self._down_queue.put(frame)
 
 
+class Observer(BaseObserver):
+    """This is a pipeline frame observer that is used as a proxy to the user
+    provided observers. That is, this is the only observer passed to the frame
+    processors. Then, every time a frame is pushed this observer will call all
+    the observers registered to the pipeline task.
+
+    """
+
+    def __init__(self, observers: List[BaseObserver] = []):
+        self._observers = observers
+
+    async def on_push_frame(
+        self,
+        src: FrameProcessor,
+        dst: FrameProcessor,
+        frame: Frame,
+        direction: FrameDirection,
+        timestamp: int,
+    ):
+        for observer in self._observers:
+            await observer.on_push_frame(src, dst, frame, direction, timestamp)
+
+
 class PipelineTask:
     def __init__(
         self,
@@ -105,6 +145,8 @@ class PipelineTask:
         self._sink = Sink(self._down_queue)
         pipeline.link(self._sink)
 
+        self._observer = Observer(params.observers)
+
     def has_finished(self):
         return self._finished
 
@@ -156,6 +198,7 @@ class PipelineTask:
             enable_metrics=self._params.enable_metrics,
             enable_usage_metrics=self._params.enable_usage_metrics,
             report_only_initial_ttfb=self._params.report_only_initial_ttfb,
+            observer=self._observer,
             clock=self._clock,
         )
         await self._source.queue_frame(start_frame, FrameDirection.DOWNSTREAM)

src/pipecat/processors/frame_processor.py

@@ -58,6 +58,7 @@ class FrameProcessor:
         self._enable_metrics = False
         self._enable_usage_metrics = False
         self._report_only_initial_ttfb = False
+        self._observer = None
 
         # Cancellation is done through CancelFrame (a system frame). This could
         # cause other events being triggered (e.g. closing a transport) which
@@ -194,6 +195,7 @@ class FrameProcessor:
             self._enable_metrics = frame.enable_metrics
             self._enable_usage_metrics = frame.enable_usage_metrics
             self._report_only_initial_ttfb = frame.report_only_initial_ttfb
+            self._observer = frame.observer
         elif isinstance(frame, StartInterruptionFrame):
             await self._start_interruption()
             await self.stop_all_metrics()
@@ -256,11 +258,20 @@ class FrameProcessor:
 
     async def __internal_push_frame(self, frame: Frame, direction: FrameDirection):
         try:
+            timestamp = self._clock.get_time()
             if direction == FrameDirection.DOWNSTREAM and self._next:
                 logger.trace(f"Pushing {frame} from {self} to {self._next}")
+                if self._observer:
+                    await self._observer.on_push_frame(
+                        self, self._next, frame, direction, timestamp
+                    )
                 await self._next.queue_frame(frame, direction)
             elif direction == FrameDirection.UPSTREAM and self._prev:
                 logger.trace(f"Pushing {frame} upstream from {self} to {self._prev}")
+                if self._observer:
+                    await self._observer.on_push_frame(
+                        self, self._prev, frame, direction, timestamp
+                    )
                 await self._prev.queue_frame(frame, direction)
         except Exception as e:
             logger.exception(f"Uncaught exception in {self}: {e}")

src/pipecat/processors/frameworks/rtvi.py

@@ -34,14 +34,15 @@ from pipecat.frames.frames import (
     InterimTranscriptionFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
+    LLMTextFrame,
     MetricsFrame,
     StartFrame,
     SystemFrame,
-    TextFrame,
     TranscriptionFrame,
     TransportMessageUrgentFrame,
     TTSStartedFrame,
     TTSStoppedFrame,
+    TTSTextFrame,
     UserStartedSpeakingFrame,
     UserStoppedSpeakingFrame,
 )
@@ -51,6 +52,7 @@ from pipecat.metrics.metrics import (
     TTFBMetricsData,
     TTSUsageMetricsData,
 )
+from pipecat.observers.base_observer import BaseObserver
 from pipecat.processors.aggregators.openai_llm_context import (
     OpenAILLMContext,
     OpenAILLMContextFrame,
@@ -479,7 +481,7 @@ class RTVIBotTranscriptionProcessor(RTVIFrameProcessor):
 
         if isinstance(frame, UserStartedSpeakingFrame):
             await self._push_aggregation()
-        elif isinstance(frame, TextFrame):
+        elif isinstance(frame, LLMTextFrame):
             self._aggregation += frame.text
             if match_endofsentence(self._aggregation):
                 await self._push_aggregation()
@@ -504,7 +506,7 @@ class RTVIBotLLMProcessor(RTVIFrameProcessor):
             await self._push_transport_message_urgent(RTVIBotLLMStartedMessage())
         elif isinstance(frame, LLMFullResponseEndFrame):
             await self._push_transport_message_urgent(RTVIBotLLMStoppedMessage())
-        elif type(frame) is TextFrame:
+        elif isinstance(frame, LLMTextFrame):
             message = RTVIBotLLMTextMessage(data=RTVITextMessageData(text=frame.text))
             await self._push_transport_message_urgent(message)
 
@@ -522,7 +524,7 @@ class RTVIBotTTSProcessor(RTVIFrameProcessor):
             await self._push_transport_message_urgent(RTVIBotTTSStartedMessage())
         elif isinstance(frame, TTSStoppedFrame):
             await self._push_transport_message_urgent(RTVIBotTTSStoppedMessage())
-        elif type(frame) is TextFrame:
+        elif isinstance(frame, TTSTextFrame):
             message = RTVIBotTTSTextMessage(data=RTVITextMessageData(text=frame.text))
             await self._push_transport_message_urgent(message)
 
@@ -563,6 +565,153 @@ class RTVIMetricsProcessor(RTVIFrameProcessor):
         await self._push_transport_message_urgent(message)
 
 
+class RTVIObserver(BaseObserver):
+    """This is a pipeline frame observer that is used to send RTVI server
+    messages to clients. The observer does not handle incoming RTVI client
+    messages, which is done by the RTVIProcessor.
+
+    """
+
+    def __init__(self, rtvi: FrameProcessor):
+        super().__init__()
+        self._rtvi = rtvi
+        self._bot_transcription = ""
+        self._frames_seen = set()
+
+    async def on_push_frame(
+        self,
+        src: FrameProcessor,
+        dst: FrameProcessor,
+        frame: Frame,
+        direction: FrameDirection,
+        timestamp: int,
+    ):
+        # If we have already seen this frame, let's skip it.
+        if frame.id in self._frames_seen:
+            return
+        self._frames_seen.add(frame.id)
+
+        if isinstance(frame, (UserStartedSpeakingFrame, UserStoppedSpeakingFrame)):
+            await self._handle_interruptions(frame)
+        elif isinstance(frame, (BotStartedSpeakingFrame, BotStoppedSpeakingFrame)):
+            await self._handle_bot_speaking(frame)
+        elif isinstance(frame, (TranscriptionFrame, InterimTranscriptionFrame)):
+            await self._handle_user_transcriptions(frame)
+        elif isinstance(frame, OpenAILLMContextFrame):
+            await self._handle_context(frame)
+        elif isinstance(frame, UserStartedSpeakingFrame):
+            await self._push_bot_transcription()
+        elif isinstance(frame, LLMFullResponseStartFrame):
+            await self._push_transport_message_urgent(RTVIBotLLMStartedMessage())
+        elif isinstance(frame, LLMFullResponseEndFrame):
+            await self._push_transport_message_urgent(RTVIBotLLMStoppedMessage())
+        elif isinstance(frame, LLMTextFrame):
+            await self._handle_llm_text_frame(frame)
+        elif isinstance(frame, TTSStartedFrame):
+            await self._push_transport_message_urgent(RTVIBotTTSStartedMessage())
+        elif isinstance(frame, TTSStoppedFrame):
+            await self._push_transport_message_urgent(RTVIBotTTSStoppedMessage())
+        elif isinstance(frame, TTSTextFrame):
+            message = RTVIBotTTSTextMessage(data=RTVITextMessageData(text=frame.text))
+            await self._push_transport_message_urgent(message)
+        elif isinstance(frame, MetricsFrame):
+            await self._handle_metrics(frame)
+
+    async def _push_transport_message_urgent(self, model: BaseModel, exclude_none: bool = True):
+        frame = TransportMessageUrgentFrame(message=model.model_dump(exclude_none=exclude_none))
+        await self._rtvi.push_frame(frame)
+
+    async def _push_bot_transcription(self):
+        if len(self._bot_transcription) > 0:
+            message = RTVIBotTranscriptionMessage(
+                data=RTVITextMessageData(text=self._bot_transcription)
+            )
+            await self._push_transport_message_urgent(message)
+            self._bot_transcription = ""
+
+    async def _handle_interruptions(self, frame: Frame):
+        message = None
+        if isinstance(frame, UserStartedSpeakingFrame):
+            message = RTVIUserStartedSpeakingMessage()
+        elif isinstance(frame, UserStoppedSpeakingFrame):
+            message = RTVIUserStoppedSpeakingMessage()
+
+        if message:
+            await self._push_transport_message_urgent(message)
+
+    async def _handle_bot_speaking(self, frame: Frame):
+        message = None
+        if isinstance(frame, BotStartedSpeakingFrame):
+            message = RTVIBotStartedSpeakingMessage()
+        elif isinstance(frame, BotStoppedSpeakingFrame):
+            message = RTVIBotStoppedSpeakingMessage()
+
+        if message:
+            await self._push_transport_message_urgent(message)
+
+    async def _handle_llm_text_frame(self, frame: LLMTextFrame):
+        message = RTVIBotLLMTextMessage(data=RTVITextMessageData(text=frame.text))
+        await self._push_transport_message_urgent(message)
+
+        self._bot_transcription += frame.text
+        if match_endofsentence(self._bot_transcription):
+            await self._push_bot_transcription()
+
+    async def _handle_user_transcriptions(self, frame: Frame):
+        message = None
+        if isinstance(frame, TranscriptionFrame):
+            message = RTVIUserTranscriptionMessage(
+                data=RTVIUserTranscriptionMessageData(
+                    text=frame.text, user_id=frame.user_id, timestamp=frame.timestamp, final=True
+                )
+            )
+        elif isinstance(frame, InterimTranscriptionFrame):
+            message = RTVIUserTranscriptionMessage(
+                data=RTVIUserTranscriptionMessageData(
+                    text=frame.text, user_id=frame.user_id, timestamp=frame.timestamp, final=False
+                )
+            )
+
+        if message:
+            await self._push_transport_message_urgent(message)
+
+    async def _handle_context(self, frame: OpenAILLMContextFrame):
+        messages = frame.context.messages
+        if len(messages) > 0:
+            message = messages[-1]
+            if message["role"] == "user":
+                content = message["content"]
+                if isinstance(content, list):
+                    text = " ".join(item["text"] for item in content if "text" in item)
+                else:
+                    text = content
+                rtvi_message = RTVIUserLLMTextMessage(data=RTVITextMessageData(text=text))
+                await self._push_transport_message_urgent(rtvi_message)
+
+    async def _handle_metrics(self, frame: MetricsFrame):
+        metrics = {}
+        for d in frame.data:
+            if isinstance(d, TTFBMetricsData):
+                if "ttfb" not in metrics:
+                    metrics["ttfb"] = []
+                metrics["ttfb"].append(d.model_dump(exclude_none=True))
+            elif isinstance(d, ProcessingMetricsData):
+                if "processing" not in metrics:
+                    metrics["processing"] = []
+                metrics["processing"].append(d.model_dump(exclude_none=True))
+            elif isinstance(d, LLMUsageMetricsData):
+                if "tokens" not in metrics:
+                    metrics["tokens"] = []
+                metrics["tokens"].append(d.value.model_dump(exclude_none=True))
+            elif isinstance(d, TTSUsageMetricsData):
+                if "characters" not in metrics:
+                    metrics["characters"] = []
+                metrics["characters"].append(d.model_dump(exclude_none=True))
+
+        message = RTVIMetricsMessage(data=metrics)
+        await self._push_transport_message_urgent(message)
+
+
 class RTVIProcessor(FrameProcessor):
     def __init__(
         self,
@@ -593,6 +742,9 @@ class RTVIProcessor(FrameProcessor):
         self._register_event_handler("on_bot_started")
         self._register_event_handler("on_client_ready")
 
+    def observer(self) -> RTVIObserver:
+        return RTVIObserver(self)
+
     def register_action(self, action: RTVIAction):
         id = self._action_id(action.service, action.action)
         self._registered_actions[id] = action

src/pipecat/services/ai_services.py

@@ -30,6 +30,7 @@ from pipecat.frames.frames import (
     TTSSpeakFrame,
     TTSStartedFrame,
     TTSStoppedFrame,
+    TTSTextFrame,
     TTSUpdateSettingsFrame,
     UserImageRequestFrame,
     VisionImageRawFrame,
@@ -358,7 +359,7 @@ class TTSService(AIService):
         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(TextFrame(text))
+            await self.push_frame(TTSTextFrame(text))
 
     async def _stop_frame_handler(self):
         try:
@@ -437,7 +438,7 @@ class WordTTSService(TTSService):
                     frame = TTSStoppedFrame()
                     frame.pts = last_pts
                 else:
-                    frame = TextFrame(word)
+                    frame = TTSTextFrame(word)
                     frame.pts = self._initial_word_timestamp + timestamp
                 if frame:
                     last_pts = frame.pts

src/pipecat/services/anthropic.py

@@ -26,10 +26,10 @@ from pipecat.frames.frames import (
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     LLMMessagesFrame,
+    LLMTextFrame,
     LLMUpdateSettingsFrame,
     OpenAILLMContextAssistantTimestampFrame,
     StartInterruptionFrame,
-    TextFrame,
     UserImageRawFrame,
     UserImageRequestFrame,
     VisionImageRawFrame,
@@ -191,7 +191,7 @@ class AnthropicLLMService(LLMService):
 
                 if event.type == "content_block_delta":
                     if hasattr(event.delta, "text"):
-                        await self.push_frame(TextFrame(event.delta.text))
+                        await self.push_frame(LLMTextFrame(event.delta.text))
                         completion_tokens_estimate += self._estimate_tokens(event.delta.text)
                     elif hasattr(event.delta, "partial_json") and tool_use_block:
                         json_accumulator += event.delta.partial_json

src/pipecat/services/gemini_multimodal_live/gemini.py

@@ -28,10 +28,10 @@ from pipecat.frames.frames import (
     LLMFullResponseStartFrame,
     LLMMessagesAppendFrame,
     LLMSetToolsFrame,
+    LLMTextFrame,
     LLMUpdateSettingsFrame,
     StartFrame,
     StartInterruptionFrame,
-    TextFrame,
     TranscriptionFrame,
     TTSAudioRawFrame,
     TTSStartedFrame,
@@ -295,7 +295,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
         # definitely feels like a hack. Need to revisit when the API evolves.
         # context.add_message({"role": "assistant", "content": [{"type": "text", "text": text}]})
         await self.push_frame(LLMFullResponseStartFrame())
-        await self.push_frame(TextFrame(text=text))
+        await self.push_frame(LLMTextFrame(text=text))
         await self.push_frame(LLMFullResponseEndFrame())
 
     async def _transcribe_audio(self, audio, context):
@@ -628,7 +628,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
                 await self.push_frame(LLMFullResponseStartFrame())
 
             self._bot_text_buffer += text
-            await self.push_frame(TextFrame(text=text))
+            await self.push_frame(LLMTextFrame(text=text))
 
         inline_data = part.inlineData
         if not inline_data:

src/pipecat/services/google.py

@@ -23,9 +23,9 @@ from pipecat.frames.frames import (
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     LLMMessagesFrame,
+    LLMTextFrame,
     LLMUpdateSettingsFrame,
     OpenAILLMContextAssistantTimestampFrame,
-    TextFrame,
     TTSAudioRawFrame,
     TTSStartedFrame,
     TTSStoppedFrame,
@@ -698,7 +698,7 @@ class GoogleLLMService(LLMService):
                 try:
                     for c in chunk.parts:
                         if c.text:
-                            await self.push_frame(TextFrame(c.text))
+                            await self.push_frame(LLMTextFrame(c.text))
                         elif c.function_call:
                             logger.debug(f"!!! Function call: {c.function_call}")
                             args = type(c.function_call).to_dict(c.function_call).get("args", {})

src/pipecat/services/openai.py

@@ -25,10 +25,10 @@ from pipecat.frames.frames import (
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     LLMMessagesFrame,
+    LLMTextFrame,
     LLMUpdateSettingsFrame,
     OpenAILLMContextAssistantTimestampFrame,
     StartInterruptionFrame,
-    TextFrame,
     TTSAudioRawFrame,
     TTSStartedFrame,
     TTSStoppedFrame,
@@ -258,7 +258,7 @@ class BaseOpenAILLMService(LLMService):
                     # Keep iterating through the response to collect all the argument fragments
                     arguments += tool_call.function.arguments
             elif chunk.choices[0].delta.content:
-                await self.push_frame(TextFrame(chunk.choices[0].delta.content))
+                await self.push_frame(LLMTextFrame(chunk.choices[0].delta.content))
 
         # if we got a function name and arguments, check to see if it's a function with
         # a registered handler. If so, run the registered callback, save the result to

src/pipecat/services/openai_realtime_beta/openai.py

@@ -24,11 +24,11 @@ from pipecat.frames.frames import (
     LLMFullResponseStartFrame,
     LLMMessagesAppendFrame,
     LLMSetToolsFrame,
+    LLMTextFrame,
     LLMUpdateSettingsFrame,
     StartFrame,
     StartInterruptionFrame,
     StopInterruptionFrame,
-    TextFrame,
     TranscriptionFrame,
     TTSAudioRawFrame,
     TTSStartedFrame,
@@ -458,7 +458,7 @@ class OpenAIRealtimeBetaLLMService(LLMService):
 
     async def _handle_evt_audio_transcript_delta(self, evt):
         if evt.delta:
-            await self.push_frame(TextFrame(evt.delta))
+            await self.push_frame(LLMTextFrame(evt.delta))
 
     async def _handle_evt_speech_started(self, evt):
         await self._truncate_current_audio_response()