Merge pull request #1166 from pipecat-ai/aleix/google-rtvi-observer

rtvi: separate specific google RTVI into a GoogleRTVIObserver

CHANGELOG.md

@@ -5,10 +5,28 @@ All notable changes to **Pipecat** will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Changed
+
+- `RTVIObserver` doesn't handle `LLMSearchResponseFrame` frames anymore. For
+  now, to handle those frames you need to create a `GoogleRTVIObserver` instead.
+
+### Deprecated
+
+- `RTVI.observer()` is now deprecated, instantiate an `RTVIObserver` directly
+  instead.
+
+- All RTVI frame processors (e.g. `RTVISpeakingProcessor`,
+  `RTVIBotLLMProcessor`) are now deprecated, instantiate an `RTVIObserver`
+  instead.
+
 ## [0.0.56] - 2025-02-06
 
 ### Changed
 
+- Use `gemini-2.0-flash-001` as the default model for `GoogleLLMSerivce`.
+
 - Improved foundational examples 22b, 22c, and 22d to support function calling.
   With these base examples, `FunctionCallInProgressFrame` and
   `FunctionCallResultFrame` will no longer be blocked by the gates.
@@ -33,10 +51,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   and should be set manually from the serializer constructor if a different
   value is needed.
 
-### Changed
-
-- Use `gemini-2.0-flash-001` as the default model for `GoogleLLMSerivce`.
-
 ### Other
 
 - Added a new `sentry-metrics` example.
@@ -119,7 +133,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - `AudioBufferProcessor.reset_audio_buffers()` has been removed, use
   `AudioBufferProcessor.start_recording()` and
-  ``AudioBufferProcessor.stop_recording()` instead.
+  `AudioBufferProcessor.stop_recording()` instead.
 
 ### Fixed
 

examples/foundational/32-gemini-grounding-metadata.py

@@ -89,6 +89,7 @@ async def main():
             api_key=os.getenv("GOOGLE_API_KEY"),
             system_instruction=system_instruction,
             tools=tools,
+            model="gemini-1.5-flash-002",
         )
 
         context = OpenAILLMContext(

examples/news-chatbot/server/news_bot.py

@@ -23,7 +23,7 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIProcessor
 from pipecat.services.cartesia import CartesiaTTSService
 from pipecat.services.deepgram import DeepgramSTTService
-from pipecat.services.google import GoogleLLMService, LLMSearchResponseFrame
+from pipecat.services.google import GoogleLLMService, GoogleRTVIObserver, LLMSearchResponseFrame
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 from pipecat.utils.text.markdown_text_filter import MarkdownTextFilter
 
@@ -102,6 +102,7 @@ async def main():
 
         llm = GoogleLLMService(
             api_key=os.getenv("GOOGLE_API_KEY"),
+            model="gemini-1.5-flash-002",
             system_instruction=system_instruction,
             tools=tools,
         )
@@ -141,7 +142,7 @@ async def main():
             pipeline,
             PipelineParams(
                 allow_interruptions=True,
-                observers=[rtvi.observer()],
+                observers=[GoogleRTVIObserver(rtvi)],
             ),
         )
 

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

@@ -40,7 +40,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 RTVIConfig, RTVIProcessor
+from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIObserver, RTVIProcessor
 from pipecat.services.gemini_multimodal_live.gemini import GeminiMultimodalLiveLLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 
@@ -176,7 +176,7 @@ async def main():
                 allow_interruptions=True,
                 enable_metrics=True,
                 enable_usage_metrics=True,
-                observers=[rtvi.observer()],
+                observers=[RTVIObserver(rtvi)],
             ),
         )
         await task.queue_frame(quiet_frame)

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

@@ -40,7 +40,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 RTVIConfig, RTVIProcessor
+from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIObserver, RTVIProcessor
 from pipecat.services.elevenlabs import ElevenLabsTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
@@ -202,7 +202,7 @@ async def main():
                 allow_interruptions=True,
                 enable_metrics=True,
                 enable_usage_metrics=True,
-                observers=[rtvi.observer()],
+                observers=[RTVIObserver(rtvi)],
             ),
         )
         await task.queue_frame(quiet_frame)

src/pipecat/processors/frameworks/rtvi.py

@@ -58,7 +58,6 @@ from pipecat.processors.aggregators.openai_llm_context import (
     OpenAILLMContextFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
-from pipecat.services.google.frames import LLMSearchOrigin, LLMSearchResponseFrame
 from pipecat.utils.string import match_endofsentence
 
 RTVI_PROTOCOL_VERSION = "0.3.0"
@@ -296,12 +295,6 @@ class RTVITextMessageData(BaseModel):
     text: str
 
 
-class RTVISearchResponseMessageData(BaseModel):
-    search_result: Optional[str]
-    rendered_content: Optional[str]
-    origins: List[LLMSearchOrigin]
-
-
 class RTVIBotTranscriptionMessage(BaseModel):
     label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL
     type: Literal["bot-transcription"] = "bot-transcription"
@@ -314,12 +307,6 @@ class RTVIBotLLMTextMessage(BaseModel):
     data: RTVITextMessageData
 
 
-class RTVIBotLLMSearchResponseMessage(BaseModel):
-    label: Literal["rtvi-ai"] = "rtvi-ai"
-    type: Literal["bot-llm-search-response"] = "bot-llm-search-response"
-    data: RTVISearchResponseMessageData
-
-
 class RTVIBotTTSTextMessage(BaseModel):
     label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL
     type: Literal["bot-tts-text"] = "bot-tts-text"
@@ -397,6 +384,15 @@ class RTVISpeakingProcessor(RTVIFrameProcessor):
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
 
+        import warnings
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "'RTVISpeakingProcessor' is deprecated, use an 'RTVIObserver' instead.",
+                DeprecationWarning,
+            )
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
 
@@ -432,6 +428,15 @@ class RTVIUserTranscriptionProcessor(RTVIFrameProcessor):
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
 
+        import warnings
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "'RTVIUserTranscriptionProcessor' is deprecated, use an 'RTVIObserver' instead.",
+                DeprecationWarning,
+            )
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
 
@@ -463,6 +468,15 @@ class RTVIUserLLMTextProcessor(RTVIFrameProcessor):
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
 
+        import warnings
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "'RTVIUserLLMTextProcessor' is deprecated, use an 'RTVIObserver' instead.",
+                DeprecationWarning,
+            )
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
 
@@ -490,6 +504,15 @@ class RTVIBotTranscriptionProcessor(RTVIFrameProcessor):
         super().__init__()
         self._aggregation = ""
 
+        import warnings
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "'RTVIBotTranscriptionProcessor' is deprecated, use an 'RTVIObserver' instead.",
+                DeprecationWarning,
+            )
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
 
@@ -513,6 +536,15 @@ class RTVIBotLLMProcessor(RTVIFrameProcessor):
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
 
+        import warnings
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "'RTVIBotLLMProcessor' is deprecated, use an 'RTVIObserver' instead.",
+                DeprecationWarning,
+            )
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
 
@@ -531,6 +563,15 @@ class RTVIBotTTSProcessor(RTVIFrameProcessor):
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
 
+        import warnings
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "'RTVIBotTTSProcessor' is deprecated, use an 'RTVIObserver' instead.",
+                DeprecationWarning,
+            )
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
 
@@ -549,6 +590,15 @@ class RTVIMetricsProcessor(RTVIFrameProcessor):
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
 
+        import warnings
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "'RTVIMetricsProcessor' is deprecated, use an 'RTVIObserver' instead.",
+                DeprecationWarning,
+            )
+
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
 
@@ -618,24 +668,22 @@ class RTVIObserver(BaseObserver):
         elif isinstance(frame, UserStartedSpeakingFrame):
             await self._push_bot_transcription()
         elif isinstance(frame, LLMFullResponseStartFrame):
-            await self._push_transport_message_urgent(RTVIBotLLMStartedMessage())
+            await self.push_transport_message_urgent(RTVIBotLLMStartedMessage())
         elif isinstance(frame, LLMFullResponseEndFrame):
-            await self._push_transport_message_urgent(RTVIBotLLMStoppedMessage())
+            await self.push_transport_message_urgent(RTVIBotLLMStoppedMessage())
         elif isinstance(frame, LLMTextFrame):
             await self._handle_llm_text_frame(frame)
-        elif isinstance(frame, LLMSearchResponseFrame):
-            await self._handle_llm_search_response_frame(frame)
         elif isinstance(frame, TTSStartedFrame):
-            await self._push_transport_message_urgent(RTVIBotTTSStartedMessage())
+            await self.push_transport_message_urgent(RTVIBotTTSStartedMessage())
         elif isinstance(frame, TTSStoppedFrame):
-            await self._push_transport_message_urgent(RTVIBotTTSStoppedMessage())
+            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)
+            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):
+    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)
 
@@ -644,7 +692,7 @@ class RTVIObserver(BaseObserver):
             message = RTVIBotTranscriptionMessage(
                 data=RTVITextMessageData(text=self._bot_transcription)
             )
-            await self._push_transport_message_urgent(message)
+            await self.push_transport_message_urgent(message)
             self._bot_transcription = ""
 
     async def _handle_interruptions(self, frame: Frame):
@@ -655,7 +703,7 @@ class RTVIObserver(BaseObserver):
             message = RTVIUserStoppedSpeakingMessage()
 
         if message:
-            await self._push_transport_message_urgent(message)
+            await self.push_transport_message_urgent(message)
 
     async def _handle_bot_speaking(self, frame: Frame):
         message = None
@@ -665,26 +713,16 @@ class RTVIObserver(BaseObserver):
             message = RTVIBotStoppedSpeakingMessage()
 
         if message:
-            await self._push_transport_message_urgent(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)
+        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_llm_search_response_frame(self, frame: LLMSearchResponseFrame):
-        message = RTVIBotLLMSearchResponseMessage(
-            data=RTVISearchResponseMessageData(
-                search_result=frame.search_result,
-                origins=frame.origins,
-                rendered_content=frame.rendered_content,
-            )
-        )
-        await self._push_transport_message_urgent(message)
-
     async def _handle_user_transcriptions(self, frame: Frame):
         message = None
         if isinstance(frame, TranscriptionFrame):
@@ -701,7 +739,7 @@ class RTVIObserver(BaseObserver):
             )
 
         if message:
-            await self._push_transport_message_urgent(message)
+            await self.push_transport_message_urgent(message)
 
     async def _handle_context(self, frame: OpenAILLMContextFrame):
         try:
@@ -715,7 +753,7 @@ class RTVIObserver(BaseObserver):
                     else:
                         text = content
                     rtvi_message = RTVIUserLLMTextMessage(data=RTVITextMessageData(text=text))
-                    await self._push_transport_message_urgent(rtvi_message)
+                    await self.push_transport_message_urgent(rtvi_message)
         except TypeError as e:
             logger.warning(f"Caught an error while trying to handle context: {e}")
 
@@ -740,7 +778,7 @@ class RTVIObserver(BaseObserver):
                 metrics["characters"].append(d.model_dump(exclude_none=True))
 
         message = RTVIMetricsMessage(data=metrics)
-        await self._push_transport_message_urgent(message)
+        await self.push_transport_message_urgent(message)
 
 
 class RTVIProcessor(FrameProcessor):
@@ -774,6 +812,15 @@ class RTVIProcessor(FrameProcessor):
         self._register_event_handler("on_client_ready")
 
     def observer(self) -> RTVIObserver:
+        import warnings
+
+        with warnings.catch_warnings():
+            warnings.simplefilter("always")
+            warnings.warn(
+                "'RTVI.observer()' is deprecated, instantiate an 'RTVIObserver' directly instead.",
+                DeprecationWarning,
+            )
+
         return RTVIObserver(self)
 
     def register_action(self, action: RTVIAction):

src/pipecat/services/google/__init__.py

@@ -1,2 +1,3 @@
 from .frames import LLMSearchResponseFrame
 from .google import *
+from .rtvi import *

src/pipecat/services/google/rtvi.py

@@ -0,0 +1,54 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+from typing import List, Literal, Optional
+
+from pydantic import BaseModel
+
+from pipecat.frames.frames import Frame
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.processors.frameworks.rtvi import RTVIObserver
+from pipecat.services.google.frames import LLMSearchOrigin, LLMSearchResponseFrame
+
+
+class RTVISearchResponseMessageData(BaseModel):
+    search_result: Optional[str]
+    rendered_content: Optional[str]
+    origins: List[LLMSearchOrigin]
+
+
+class RTVIBotLLMSearchResponseMessage(BaseModel):
+    label: Literal["rtvi-ai"] = "rtvi-ai"
+    type: Literal["bot-llm-search-response"] = "bot-llm-search-response"
+    data: RTVISearchResponseMessageData
+
+
+class GoogleRTVIObserver(RTVIObserver):
+    def __init__(self, rtvi: FrameProcessor):
+        super().__init__(rtvi)
+
+    async def on_push_frame(
+        self,
+        src: FrameProcessor,
+        dst: FrameProcessor,
+        frame: Frame,
+        direction: FrameDirection,
+        timestamp: int,
+    ):
+        await super().on_push_frame(src, dst, frame, direction, timestamp)
+
+        if isinstance(frame, LLMSearchResponseFrame):
+            await self._handle_llm_search_response_frame(frame)
+
+    async def _handle_llm_search_response_frame(self, frame: LLMSearchResponseFrame):
+        message = RTVIBotLLMSearchResponseMessage(
+            data=RTVISearchResponseMessageData(
+                search_result=frame.search_result,
+                origins=frame.origins,
+                rendered_content=frame.rendered_content,
+            )
+        )
+        await self.push_transport_message_urgent(message)