processors(realtime-ai): add support for updating config (model, voice...)

src/pipecat/frames/frames.py

@@ -345,3 +345,17 @@ class UserImageRequestFrame(ControlFrame):
 
     def __str__(self):
         return f"{self.name}, user: {self.user_id}"
+
+
+@dataclass
+class LLMModelUpdateFrame(ControlFrame):
+    """A control frame containing a request to update to a new LLM model.
+    """
+    model: str
+
+
+@dataclass
+class TTSVoiceUpdateFrame(ControlFrame):
+    """A control frame containing a request to update to a new TTS voice.
+    """
+    voice: str

src/pipecat/processors/frameworks/realtimeai.py

@@ -9,9 +9,16 @@ import dataclasses
 from typing import List, Literal, Optional, Type
 from pydantic import BaseModel, ValidationError
 
-from pipecat.frames.frames import Frame, LLMMessagesFrame, LLMMessagesUpdateFrame, StartFrame, TransportMessageFrame
+from pipecat.frames.frames import (
+    Frame,
+    LLMMessagesUpdateFrame,
+    LLMModelUpdateFrame,
+    StartFrame,
+    TTSVoiceUpdateFrame,
+    TransportMessageFrame)
 from pipecat.pipeline.pipeline import Pipeline
-from pipecat.processors.aggregators.llm_response import LLMAssistantResponseAggregator, LLMUserResponseAggregator
+from pipecat.processors.aggregators.llm_response import (
+    LLMAssistantResponseAggregator, LLMUserResponseAggregator)
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.services.ai_services import AIService
 from pipecat.services.cartesia import CartesiaTTSService
@@ -103,12 +110,12 @@ class RealtimeAIProcessor(FrameProcessor):
             await self._send_response("setup", False, f"invalid message: {e}")
             return
 
-        print(message)
-
         try:
             match message.type:
                 case "setup":
                     await self._handle_setup(RealtimeAISetup.model_validate(message.data.setup))
+                case "config-update":
+                    await self._handle_config_update(RealtimeAIConfig.model_validate(message.data.config))
                 case "llm-get-context":
                     await self._handle_llm_get_context()
                 case "llm-update-context":
@@ -152,6 +159,17 @@ class RealtimeAIProcessor(FrameProcessor):
         except Exception as e:
             await self._send_response("setup", False, f"unable to create pipeline: {e}")
 
+    async def _handle_config_update(self, config: RealtimeAIConfig):
+        if config.llm and config.llm.model:
+            frame = LLMModelUpdateFrame(config.llm.model)
+            await self.push_frame(frame)
+        if config.llm and config.llm.messages:
+            frame = LLMMessagesUpdateFrame(config.llm.messages)
+            await self.push_frame(frame)
+        if config.tts and config.tts.voice:
+            frame = TTSVoiceUpdateFrame(config.tts.voice)
+            await self.push_frame(frame)
+
     async def _handle_llm_get_context(self):
         messages = self._tma_in.messages
         response = RealtimeAILLMContextResponse(messages=messages)

src/pipecat/services/ai_services.py

@@ -21,6 +21,7 @@ from pipecat.frames.frames import (
     StartInterruptionFrame,
     TTSStartedFrame,
     TTSStoppedFrame,
+    TTSVoiceUpdateFrame,
     TextFrame,
     VisionImageRawFrame,
 )
@@ -148,6 +149,10 @@ class TTSService(AIService):
         self._push_text_frames: bool = push_text_frames
         self._current_sentence: str = ""
 
+    @abstractmethod
+    async def set_voice(self, voice: str):
+        pass
+
     # Converts the text to audio.
     @abstractmethod
     async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
@@ -203,6 +208,8 @@ class TTSService(AIService):
                     await self.push_frame(frame, direction)
             else:
                 await self.push_frame(frame, direction)
+        elif isinstance(frame, TTSVoiceUpdateFrame):
+            await self.set_voice(frame.voice)
         else:
             await self.push_frame(frame, direction)
 

src/pipecat/services/anthropic.py

@@ -8,6 +8,7 @@ import base64
 
 from pipecat.frames.frames import (
     Frame,
+    LLMModelUpdateFrame,
     TextFrame,
     VisionImageRawFrame,
     LLMMessagesFrame,
@@ -134,6 +135,9 @@ class AnthropicLLMService(LLMService):
             context = OpenAILLMContext.from_messages(frame.messages)
         elif isinstance(frame, VisionImageRawFrame):
             context = OpenAILLMContext.from_image_frame(frame)
+        elif isinstance(frame, LLMModelUpdateFrame):
+            logger.debug(f"Switching LLM model to: [{frame.model}]")
+            self._model = frame.model
         else:
             await self.push_frame(frame, direction)
 

src/pipecat/services/azure.py

@@ -81,6 +81,10 @@ class AzureTTSService(TTSService):
     def can_generate_metrics(self) -> bool:
         return True
 
+    async def set_voice(self, voice: str):
+        logger.debug(f"Switching TTS voice to: [{voice}]")
+        self._voice = voice
+
     async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
         logger.debug(f"Generating TTS: [{text}]")
 

src/pipecat/services/cartesia.py

@@ -87,6 +87,10 @@ class CartesiaTTSService(TTSService):
     def can_generate_metrics(self) -> bool:
         return True
 
+    async def set_voice(self, voice: str):
+        logger.debug(f"Switching TTS voice to: [{voice}]")
+        self._voice_id = voice
+
     async def start(self, frame: StartFrame):
         await super().start(frame)
         await self._connect()

src/pipecat/services/deepgram.py

@@ -59,6 +59,10 @@ class DeepgramTTSService(TTSService):
     def can_generate_metrics(self) -> bool:
         return True
 
+    async def set_voice(self, voice: str):
+        logger.debug(f"Switching TTS voice to: [{voice}]")
+        self._voice = voice
+
     async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
         logger.debug(f"Generating TTS: [{text}]")
 

src/pipecat/services/elevenlabs.py

@@ -34,6 +34,10 @@ class ElevenLabsTTSService(TTSService):
     def can_generate_metrics(self) -> bool:
         return True
 
+    async def set_voice(self, voice: str):
+        logger.debug(f"Switching TTS voice to: [{voice}]")
+        self._voice_id = voice
+
     async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
         logger.debug(f"Generating TTS: [{text}]")
 

src/pipecat/services/google.py

@@ -10,6 +10,7 @@ from typing import List
 
 from pipecat.frames.frames import (
     Frame,
+    LLMModelUpdateFrame,
     TextFrame,
     VisionImageRawFrame,
     LLMMessagesFrame,
@@ -43,11 +44,14 @@ class GoogleLLMService(LLMService):
     def __init__(self, *, api_key: str, model: str = "gemini-1.5-flash-latest", **kwargs):
         super().__init__(**kwargs)
         gai.configure(api_key=api_key)
-        self._client = gai.GenerativeModel(model)
+        self._create_client(model)
 
     def can_generate_metrics(self) -> bool:
         return True
 
+    def _create_client(self, model: str):
+        self._client = gai.GenerativeModel(model)
+
     def _get_messages_from_openai_context(
             self, context: OpenAILLMContext) -> List[glm.Content]:
         openai_messages = context.get_messages()
@@ -118,6 +122,9 @@ class GoogleLLMService(LLMService):
             context = OpenAILLMContext.from_messages(frame.messages)
         elif isinstance(frame, VisionImageRawFrame):
             context = OpenAILLMContext.from_image_frame(frame)
+        elif isinstance(frame, LLMModelUpdateFrame):
+            logger.debug(f"Switching LLM model to: [{frame.model}]")
+            self._create_client(frame.model)
         else:
             await self.push_frame(frame, direction)
 

src/pipecat/services/openai.py

@@ -22,6 +22,7 @@ from pipecat.frames.frames import (
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     LLMMessagesFrame,
+    LLMModelUpdateFrame,
     TextFrame,
     URLImageRawFrame,
     VisionImageRawFrame
@@ -227,6 +228,9 @@ class BaseOpenAILLMService(LLMService):
             context = OpenAILLMContext.from_messages(frame.messages)
         elif isinstance(frame, VisionImageRawFrame):
             context = OpenAILLMContext.from_image_frame(frame)
+        elif isinstance(frame, LLMModelUpdateFrame):
+            logger.debug(f"Switching LLM model to: [{frame.model}]")
+            self._model = frame.model
         else:
             await self.push_frame(frame, direction)
 
@@ -313,6 +317,10 @@ class OpenAITTSService(TTSService):
     def can_generate_metrics(self) -> bool:
         return True
 
+    async def set_voice(self, voice: str):
+        logger.debug(f"Switching TTS voice to: [{voice}]")
+        self._voice = voice
+
     async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
         logger.debug(f"Generating TTS: [{text}]")
 

src/pipecat/services/xtts.py

@@ -54,6 +54,10 @@ class XTTSService(TTSService):
     def can_generate_metrics(self) -> bool:
         return True
 
+    async def set_voice(self, voice: str):
+        logger.debug(f"Switching TTS voice to: [{voice}]")
+        self._voice_id = voice
+
     async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
         logger.debug(f"Generating TTS: [{text}]")
         embeddings = self._studio_speakers[self._voice_id]