Merge pull request #3276 from pipecat-ai/mb/grok-realtime-cleanup

GrokRealtimeLLMService cleanup

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

@@ -205,7 +205,9 @@ class SessionProperties(BaseModel):
     Parameters:
         instructions: System instructions for the assistant.
         voice: The voice the model uses to respond. Options: Ara, Rex, Sal, Eve, Leo.
-        turn_detection: Configuration for turn detection, or None for manual.
+            Defaults to "Ara".
+        turn_detection: Configuration for turn detection. Defaults to server-side VAD.
+            Set to None for manual turn detection.
         audio: Configuration for input and output audio.
         tools: Available tools for the assistant (web_search, x_search, file_search, function).
     """
@@ -215,7 +217,9 @@ class SessionProperties(BaseModel):
 
     instructions: Optional[str] = None
     voice: Optional[GrokVoice] = "Ara"
-    turn_detection: Optional[TurnDetection] = None
+    turn_detection: Optional[TurnDetection] = Field(
+        default_factory=lambda: TurnDetection(type="server_vad")
+    )
     audio: Optional[AudioConfiguration] = None
     # Tools can be ToolsSchema when provided by user, or list of dicts for API
     tools: Optional[ToolsSchema | List[GrokTool]] = None
@@ -629,6 +633,26 @@ class ResponseAudioDone(ServerEvent):
     item_id: str
 
 
+class ResponseFunctionCallArgumentsDelta(ServerEvent):
+    """Event containing incremental function call arguments.
+
+    Parameters:
+        type: Event type, always "response.function_call_arguments.delta".
+        response_id: ID of the response.
+        item_id: ID of the conversation item.
+        call_id: ID of the function call.
+        delta: Incremental function arguments as JSON.
+        previous_item_id: ID of the previous item, if any.
+    """
+
+    type: Literal["response.function_call_arguments.delta"]
+    response_id: Optional[str] = None
+    item_id: Optional[str] = None
+    call_id: str
+    delta: str
+    previous_item_id: Optional[str] = None
+
+
 class ResponseFunctionCallArgumentsDone(ServerEvent):
     """Event indicating function call arguments are complete.
 
@@ -820,6 +844,7 @@ _server_event_types = {
     "response.output_audio_transcript.done": ResponseAudioTranscriptDone,
     "response.output_audio.delta": ResponseAudioDelta,
     "response.output_audio.done": ResponseAudioDone,
+    "response.function_call_arguments.delta": ResponseFunctionCallArgumentsDelta,
     "response.function_call_arguments.done": ResponseFunctionCallArgumentsDone,
     "response.done": ResponseDone,
 }

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

@@ -27,14 +27,12 @@ from pipecat.frames.frames import (
     EndFrame,
     Frame,
     InputAudioRawFrame,
-    InterimTranscriptionFrame,
     InterruptionFrame,
     LLMContextFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     LLMMessagesAppendFrame,
     LLMSetToolsFrame,
-    LLMTextFrame,
     LLMUpdateSettingsFrame,
     StartFrame,
     TranscriptionFrame,
@@ -57,7 +55,6 @@ 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.transcriptions.language import Language
 from pipecat.utils.time import time_now_iso8601
 
 from . import events
@@ -110,50 +107,36 @@ class GrokRealtimeLLMService(LLMService):
         self,
         *,
         api_key: str,
-        voice: events.GrokVoice = "Ara",
         base_url: str = "wss://api.x.ai/v1/realtime",
         session_properties: Optional[events.SessionProperties] = None,
         start_audio_paused: bool = False,
-        sample_rate: int = 24000,
         **kwargs,
     ):
         """Initialize the Grok Realtime Voice Agent LLM service.
 
         Args:
             api_key: xAI API key for authentication.
-            voice: Voice to use for responses. Options: Ara, Rex, Sal, Eve, Leo.
-                Defaults to "Ara".
             base_url: WebSocket base URL for the realtime API.
                 Defaults to "wss://api.x.ai/v1/realtime".
             session_properties: Configuration properties for the realtime session.
-                If None, uses default SessionProperties with the specified voice.
+                If None, uses default SessionProperties with voice "Ara".
+                To set a different voice, configure it in session_properties:
+
+                    session_properties = events.SessionProperties(voice="Rex")
+
+                Available voices: Ara, Rex, Sal, Eve, Leo.
             start_audio_paused: Whether to start with audio input paused. Defaults to False.
-            sample_rate: Audio sample rate in Hz. Supported: 8000, 16000, 21050, 24000,
-                32000, 44100, 48000. Defaults to 24000.
             **kwargs: Additional arguments passed to parent LLMService.
         """
         super().__init__(base_url=base_url, **kwargs)
 
         self.api_key = api_key
         self.base_url = base_url
-        self._sample_rate = sample_rate
-        self._voice = voice
 
-        # Initialize session_properties with voice and audio config
-        if session_properties:
-            self._session_properties = session_properties
-            # Ensure voice is set
-            if not self._session_properties.voice:
-                self._session_properties.voice = voice
-        else:
-            self._session_properties = events.SessionProperties(
-                voice=voice,
-                turn_detection=events.TurnDetection(type="server_vad"),
-                audio=events.AudioConfiguration(
-                    input=events.AudioInput(format=events.PCMAudioFormat(rate=sample_rate)),
-                    output=events.AudioOutput(format=events.PCMAudioFormat(rate=sample_rate)),
-                ),
-            )
+        # Initialize session_properties
+        self._session_properties: events.SessionProperties = (
+            session_properties or events.SessionProperties()
+        )
 
         self._audio_input_paused = start_audio_paused
         self._websocket = None
@@ -192,6 +175,50 @@ class GrokRealtimeLLMService(LLMService):
         """
         self._audio_input_paused = paused
 
+    def _get_configured_sample_rate(self, direction: str) -> Optional[int]:
+        """Get manually configured sample rate for input or output.
+
+        Args:
+            direction: Either "input" or "output".
+
+        Returns:
+            Configured sample rate or None if not manually configured.
+            For PCMU/PCMA formats, returns 8000 Hz (G.711 standard).
+        """
+        if not self._session_properties.audio:
+            return None
+
+        audio_config = (
+            self._session_properties.audio.input
+            if direction == "input"
+            else self._session_properties.audio.output
+        )
+
+        if audio_config and audio_config.format:
+            # PCM format has configurable rate
+            if hasattr(audio_config.format, "rate"):
+                return audio_config.format.rate
+            # PCMU/PCMA formats are fixed at 8000 Hz (G.711 standard)
+            elif audio_config.format.type in ("audio/pcmu", "audio/pcma"):
+                return 8000
+
+        return None
+
+    def _get_output_sample_rate(self) -> int:
+        """Get the output sample rate from session properties.
+
+        Returns:
+            Output sample rate in Hz.
+
+        Note:
+            This assumes start() has been called, which guarantees
+            session_properties.audio.output exists.
+        """
+        rate = self._get_configured_sample_rate("output")
+        if rate is None:
+            raise RuntimeError("Output sample rate not configured.")
+        return rate
+
     def _is_turn_detection_enabled(self) -> bool:
         """Check if server-side VAD is enabled."""
         if self._session_properties.turn_detection:
@@ -230,7 +257,7 @@ class GrokRealtimeLLMService(LLMService):
     ) -> int:
         """Calculate audio duration in milliseconds based on PCM audio parameters."""
         if sample_rate is None:
-            sample_rate = self._sample_rate
+            sample_rate = self._get_output_sample_rate()
         samples = total_bytes / bytes_per_sample
         duration_seconds = samples / sample_rate
         return int(duration_seconds * 1000)
@@ -260,6 +287,23 @@ 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._session_properties.audio:
+            self._session_properties.audio = events.AudioConfiguration()
+
+        # Fill in missing input configuration
+        if not self._session_properties.audio.input:
+            self._session_properties.audio.input = events.AudioInput(
+                format=events.PCMAudioFormat(rate=frame.audio_in_sample_rate)
+            )
+
+        # Fill in missing output configuration
+        if not self._session_properties.audio.output:
+            self._session_properties.audio.output = events.AudioOutput(
+                format=events.PCMAudioFormat(rate=frame.audio_out_sample_rate)
+            )
+
         await self._connect()
 
     async def stop(self, frame: EndFrame):
@@ -501,7 +545,7 @@ class GrokRealtimeLLMService(LLMService):
 
         frame = TTSAudioRawFrame(
             audio=audio,
-            sample_rate=self._sample_rate,
+            sample_rate=self._get_output_sample_rate(),
             num_channels=1,
         )
         await self.push_frame(frame)