diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 7ea04ecf2..ee25af839 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -1413,6 +1413,18 @@ class RTVIProcessor(FrameProcessor): self._registered_services[service.name] = service + def create_rtvi_observer(self, *, params: Optional[RTVIObserverParams] = None, **kwargs): + """Creates a new RTVI Observer. + + Args: + params: Settings to enable/disable specific messages. + **kwargs: Additional arguments passed to the observer. + + Returns: + A new RTVI observer. + """ + return RTVIObserver(self, params=params, **kwargs) + async def set_client_ready(self): """Mark the client as ready and trigger the ready event.""" self._client_ready = True diff --git a/src/pipecat/services/google/rtvi.py b/src/pipecat/services/google/rtvi.py index 1ef70d67d..738b0ab9d 100644 --- a/src/pipecat/services/google/rtvi.py +++ b/src/pipecat/services/google/rtvi.py @@ -4,7 +4,7 @@ # SPDX-License-Identifier: BSD 2-Clause License # -"""Google RTVI integration models and observer implementation. +"""Google RTVI processor and observer implementation. This module provides integration with Google's services through the RTVI framework, including models for search responses and an observer for handling Google-specific @@ -16,7 +16,7 @@ from typing import List, Literal, Optional from pydantic import BaseModel from pipecat.observers.base_observer import FramePushed -from pipecat.processors.frameworks.rtvi import RTVIObserver, RTVIProcessor +from pipecat.processors.frameworks.rtvi import RTVIObserver, RTVIObserverParams, RTVIProcessor from pipecat.services.google.frames import LLMSearchOrigin, LLMSearchResponseFrame @@ -86,4 +86,23 @@ class GoogleRTVIObserver(RTVIObserver): rendered_content=frame.rendered_content, ) ) - await self.push_transport_message_urgent(message) + await self.send_rtvi_message(message) + + +class GoogleRTVIProcessor(RTVIProcessor): + """RTVI processor for Google service integration. + + Creates a specific Google RTVI Observer. + """ + + def create_rtvi_observer(self, *, params: Optional[RTVIObserverParams] = None, **kwargs): + """Creates a new RTVI Observer. + + Args: + params: Settings to enable/disable specific messages. + **kwargs: Additional arguments passed to the observer. + + Returns: + A new RTVI observer. + """ + return GoogleRTVIObserver(self)