From c72c3025f6608a8d7785415bf86d79e32a7288d5 Mon Sep 17 00:00:00 2001
From: Mert Sefa AKGUN <git@msakg.com>
Date: Mon, 6 Jan 2025 18:07:59 +0300
Subject: [PATCH] feat(gemini): add configuration methods for response
 modalities

- Introduce `set_model_only_audio` and `set_model_only_text` methods
  to toggle between audio-only and text-only response modes in
  `GeminiMultimodalLiveLLMService`.
- Refactor configuration setup to a class attribute for improved
  reusability and maintenance.
- Remove redundant configuration instantiation in the WebSocket
  connection setup process.
---
 .../services/gemini_multimodal_live/gemini.py | 59 +++++++++++--------
 1 file changed, 35 insertions(+), 24 deletions(-)

diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py
index dea63b3c8..687c63c0b 100644
--- a/src/pipecat/services/gemini_multimodal_live/gemini.py
+++ b/src/pipecat/services/gemini_multimodal_live/gemini.py
@@ -199,6 +199,28 @@ class GeminiMultimodalLiveLLMService(LLMService):
             "extra": params.extra if isinstance(params.extra, dict) else {},
         }
 
+        self.config = events.Config.model_validate(
+            {
+                "setup": {
+                    "model": self._model_name,
+                    "generation_config": {
+                        "frequency_penalty": self._settings["frequency_penalty"],
+                        "max_output_tokens": self._settings["max_tokens"],  # Not supported yet
+                        "presence_penalty": self._settings["presence_penalty"],
+                        "temperature": self._settings["temperature"],
+                        "top_k": self._settings["top_k"],
+                        "top_p": self._settings["top_p"],
+                        "response_modalities": ["AUDIO"],
+                        "speech_config": {
+                            "voice_config": {
+                                "prebuilt_voice_config": {"voice_name": self._voice_id}
+                            },
+                        },
+                    },
+                },
+            }
+        )
+
     def can_generate_metrics(self) -> bool:
         return True
 
@@ -208,6 +230,16 @@ class GeminiMultimodalLiveLLMService(LLMService):
     def set_video_input_paused(self, paused: bool):
         self._video_input_paused = paused
 
+    def set_model_only_audio(self):
+        self.config.setup.generation_config["response_modalities"] = ["AUDIO"]
+        self.config.setup.generation_config["speech_config"] = {
+            "voice_config": {"prebuilt_voice_config": {"voice_name": self._voice_id}}
+        }
+
+    def set_model_only_text(self):
+        self.config.setup.generation_config["response_modalities"] = ["TEXT"]
+        self.config.setup.generation_config["speech_config"] = None
+
     async def set_context(self, context: OpenAILLMContext):
         """Set the context explicitly from outside the pipeline.
 
@@ -372,39 +404,18 @@ class GeminiMultimodalLiveLLMService(LLMService):
             logger.info(f"Connecting to {uri}")
             self._websocket = await websockets.connect(uri=uri)
             self._receive_task = self.get_event_loop().create_task(self._receive_task_handler())
-            config = events.Config.model_validate(
-                {
-                    "setup": {
-                        "model": self._model_name,
-                        "generation_config": {
-                            "frequency_penalty": self._settings["frequency_penalty"],
-                            "max_output_tokens": self._settings["max_tokens"],  # Not supported yet
-                            "presence_penalty": self._settings["presence_penalty"],
-                            "temperature": self._settings["temperature"],
-                            "top_k": self._settings["top_k"],
-                            "top_p": self._settings["top_p"],
-                            "response_modalities": ["AUDIO"],
-                            "speech_config": {
-                                "voice_config": {
-                                    "prebuilt_voice_config": {"voice_name": self._voice_id}
-                                },
-                            },
-                        },
-                    },
-                }
-            )
 
             system_instruction = self._system_instruction or ""
             if self._context and hasattr(self._context, "extract_system_instructions"):
                 system_instruction += "\n" + self._context.extract_system_instructions()
             if system_instruction:
                 logger.debug(f"Setting system instruction: {system_instruction}")
-                config.setup.system_instruction = events.SystemInstruction(
+                self.config.setup.system_instruction = events.SystemInstruction(
                     parts=[events.ContentPart(text=system_instruction)]
                 )
             if self._tools:
-                config.setup.tools = self._tools
-            await self.send_client_event(config)
+                self.config.setup.tools = self._tools
+            await self.send_client_event(self.config)
 
         except Exception as e:
             logger.error(f"{self} initialization error: {e}")