diff --git a/CHANGELOG.md b/CHANGELOG.md index 6b4296cc2..7f7cbbaa7 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/examples/foundational/31-gemini-grounding-metadata.py b/examples/foundational/32-gemini-grounding-metadata.py similarity index 99% rename from examples/foundational/31-gemini-grounding-metadata.py rename to examples/foundational/32-gemini-grounding-metadata.py index f82e9029c..67694652f 100644 --- a/examples/foundational/31-gemini-grounding-metadata.py +++ b/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( diff --git a/examples/news-chatbot/server/news_bot.py b/examples/news-chatbot/server/news_bot.py index f1c83c2fd..7aed4c3b9 100644 --- a/examples/news-chatbot/server/news_bot.py +++ b/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)], ), ) diff --git a/examples/simple-chatbot/server/bot-gemini.py b/examples/simple-chatbot/server/bot-gemini.py index a7e7717ce..ecf38f159 100644 --- a/examples/simple-chatbot/server/bot-gemini.py +++ b/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) diff --git a/examples/simple-chatbot/server/bot-openai.py b/examples/simple-chatbot/server/bot-openai.py index 0d30a5d87..51c421c7d 100644 --- a/examples/simple-chatbot/server/bot-openai.py +++ b/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) diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 078b76056..b5965c64b 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/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): diff --git a/src/pipecat/services/google/__init__.py b/src/pipecat/services/google/__init__.py index 1b81980e6..3e63a3ba9 100644 --- a/src/pipecat/services/google/__init__.py +++ b/src/pipecat/services/google/__init__.py @@ -1,2 +1,3 @@ from .frames import LLMSearchResponseFrame from .google import * +from .rtvi import * diff --git a/src/pipecat/services/google/rtvi.py b/src/pipecat/services/google/rtvi.py new file mode 100644 index 000000000..88e67e6c6 --- /dev/null +++ b/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)