Fix Grok Realtime dynamic session properties updating, and update corresponding 55zo example

examples/foundational/55zl-update-settings-openai-realtime.py

@@ -22,11 +22,11 @@ from pipecat.processors.aggregators.llm_response_universal import (
 )
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
+from pipecat.services.openai.realtime import events
 from pipecat.services.openai.realtime.llm import (
     OpenAIRealtimeLLMService,
     OpenAIRealtimeLLMSettings,
 )
-from pipecat.services.openai_realtime import events
 from pipecat.transports.base_transport import BaseTransport, TransportParams
 from pipecat.transports.daily.transport import DailyParams
 from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams

examples/foundational/55zo-update-settings-grok-realtime.py

@@ -16,9 +16,13 @@ from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
 from pipecat.processors.aggregators.llm_context import LLMContext
-from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
+from pipecat.processors.aggregators.llm_response_universal import (
+    AssistantTurnStoppedMessage,
+    LLMContextAggregatorPair,
+)
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
+from pipecat.services.grok.realtime import events
 from pipecat.services.grok.realtime.llm import (
     GrokRealtimeLLMService,
     GrokRealtimeLLMSettings,
@@ -51,7 +55,7 @@ transport_params = {
 async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     logger.info(f"Starting bot")
 
-    llm = GrokRealtimeLLMService(api_key=os.getenv("XAI_API_KEY"))
+    llm = GrokRealtimeLLMService(api_key=os.getenv("GROK_API_KEY"))
 
     messages = [
         {
@@ -82,15 +86,25 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
     )
 
+    @assistant_aggregator.event_handler("on_assistant_turn_stopped")
+    async def on_assistant_turn_stopped(aggregator, message: AssistantTurnStoppedMessage):
+        timestamp = f"[{message.timestamp}] " if message.timestamp else ""
+        line = f"{timestamp}assistant: {message.content}"
+        logger.info(f"Transcript: {line}")
+
     @transport.event_handler("on_client_connected")
     async def on_client_connected(transport, client):
         logger.info(f"Client connected")
         await task.queue_frames([LLMRunFrame()])
 
         await asyncio.sleep(10)
-        logger.info("Updating Grok Realtime LLM settings: temperature=0.1")
+        logger.info("Updating Grok Realtime LLM settings: voice='Rex'")
         await task.queue_frame(
-            LLMUpdateSettingsFrame(update=GrokRealtimeLLMSettings(temperature=0.1))
+            LLMUpdateSettingsFrame(
+                update=GrokRealtimeLLMSettings(
+                    session_properties=events.SessionProperties(voice="Rex")
+                )
+            )
         )
 
     @transport.event_handler("on_client_disconnected")

src/pipecat/services/grok/realtime/llm.py

@@ -56,7 +56,7 @@ from pipecat.processors.aggregators.llm_response_universal import (
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.llm_service import FunctionCallFromLLM, LLMService
-from pipecat.services.settings import NOT_GIVEN, LLMSettings
+from pipecat.services.settings import NOT_GIVEN, LLMSettings, _NotGiven
 from pipecat.utils.time import time_now_iso8601
 
 from . import events
@@ -94,7 +94,9 @@ class GrokRealtimeLLMSettings(LLMSettings):
         session_properties: Grok Realtime session configuration.
     """
 
-    session_properties: Any = field(default_factory=lambda: NOT_GIVEN)
+    session_properties: events.SessionProperties | _NotGiven = field(
+        default_factory=lambda: NOT_GIVEN
+    )
 
 
 class GrokRealtimeLLMService(LLMService):
@@ -294,6 +296,27 @@ class GrokRealtimeLLMService(LLMService):
     # Standard AIService frame handling
     #
 
+    def _ensure_audio_config(self, input_sample_rate: int, output_sample_rate: int):
+        """Ensure session_properties.audio has input and output configs.
+
+        Fills in any missing audio configuration using the given sample rates.
+
+        Args:
+            input_sample_rate: Sample rate for audio input (Hz).
+            output_sample_rate: Sample rate for audio output (Hz).
+        """
+        props = self._settings.session_properties
+        if not props.audio:
+            props.audio = events.AudioConfiguration()
+        if not props.audio.input:
+            props.audio.input = events.AudioInput(
+                format=events.PCMAudioFormat(rate=input_sample_rate)
+            )
+        if not props.audio.output:
+            props.audio.output = events.AudioOutput(
+                format=events.PCMAudioFormat(rate=output_sample_rate)
+            )
+
     async def start(self, frame: StartFrame):
         """Start the service and establish WebSocket connection.
 
@@ -301,23 +324,7 @@ class GrokRealtimeLLMService(LLMService):
             frame: The start frame triggering service initialization.
         """
         await super().start(frame)
-
-        # Ensure audio configuration exists with both input and output
-        if not self._settings.session_properties.audio:
-            self._settings.session_properties.audio = events.AudioConfiguration()
-
-        # Fill in missing input configuration
-        if not self._settings.session_properties.audio.input:
-            self._settings.session_properties.audio.input = events.AudioInput(
-                format=events.PCMAudioFormat(rate=frame.audio_in_sample_rate)
-            )
-
-        # Fill in missing output configuration
-        if not self._settings.session_properties.audio.output:
-            self._settings.session_properties.audio.output = events.AudioOutput(
-                format=events.PCMAudioFormat(rate=frame.audio_out_sample_rate)
-            )
-
+        self._ensure_audio_config(frame.audio_in_sample_rate, frame.audio_out_sample_rate)
         await self._connect()
 
     async def stop(self, frame: EndFrame):
@@ -458,9 +465,23 @@ class GrokRealtimeLLMService(LLMService):
 
     async def _update_settings(self, update):
         """Apply a settings update, sending a session update if needed."""
+        # Capture current sample rates before the update replaces them.
+        input_rate = self._get_configured_sample_rate("input")
+        output_rate = self._get_configured_sample_rate("output")
+
         changed = await super()._update_settings(update)
+
         if "session_properties" in changed:
+            if input_rate and output_rate:
+                self._ensure_audio_config(input_rate, output_rate)
+            else:
+                logger.warning(
+                    "Attempting to apply session properties update without configured sample rates. "
+                    "Audio configuration may be incomplete."
+                )
             await self._send_session_update()
+
+        self._warn_unhandled_updated_settings(changed.keys() - {"session_properties"})
         return changed
 
     async def _send_session_update(self):