diff --git a/CHANGELOG.md b/CHANGELOG.md index 6b4296cc2..51f68faf8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,10 +5,19 @@ 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. + ## [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 +42,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 +124,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/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 078b76056..630f0e1d6 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" @@ -618,24 +605,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 +629,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 +640,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 +650,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 +676,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 +690,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 +715,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): diff --git a/src/pipecat/services/google/rtvi.py b/src/pipecat/services/google/rtvi.py new file mode 100644 index 000000000..4d9860cfa --- /dev/null +++ b/src/pipecat/services/google/rtvi.py @@ -0,0 +1,51 @@ +# +# 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): + 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)