feat(openai): add xAI inference realtime defaults

livekit-plugins/livekit-plugins-openai/livekit/plugins/openai/realtime/inference_realtime_model.py

@@ -19,20 +19,32 @@
     APIConnectOptions,
     NotGivenOr,
 )
+from livekit.agents.utils.misc import is_given
 from openai.types.realtime import (
     AudioTranscription,
     NoiseReductionType,
     RealtimeAudioInputTurnDetection,
     RealtimeReasoning,
 )
 from openai.types.realtime.realtime_audio_config_input import NoiseReduction
+from openai.types.realtime.realtime_audio_input_turn_detection import ServerVad
 from openai.types.realtime.realtime_session_create_response import Tracing
 from openai.types.realtime.realtime_truncation import RealtimeTruncation
 
 from .realtime_model import DEFAULT_VOICE, RealtimeModel, RealtimeSession
 
 InferenceClass = Literal["priority", "standard", "low"]
 
+_XAI_DEFAULT_INPUT_AUDIO_TRANSCRIPTION = AudioTranscription(model="grok-transcribe")
+_XAI_DEFAULT_TURN_DETECTION = ServerVad(
+    type="server_vad",
+    threshold=0.5,
+    prefix_padding_ms=300,
+    silence_duration_ms=200,
+    create_response=True,
+    interrupt_response=True,
+)
+
 
 @dataclass
 class _InferenceOptions:
@@ -54,7 +66,7 @@ def __init__(
         api_key: str | None = None,
         api_secret: str | None = None,
         inference_class: InferenceClass | None = None,
-        voice: str = DEFAULT_VOICE,
+        voice: NotGivenOr[str] = NOT_GIVEN,
         modalities: NotGivenOr[list[Literal["text", "audio"]]] = NOT_GIVEN,
         input_audio_transcription: NotGivenOr[AudioTranscription | None] = NOT_GIVEN,
         input_audio_noise_reduction: NotGivenOr[
@@ -89,13 +101,23 @@ def __init__(
                 "api_secret is required, either as argument or set LIVEKIT_API_SECRET environmental variable"
             )
 
+        is_xai = model.startswith("xai/")
+        resolved_voice = voice if is_given(voice) else "eve" if is_xai else DEFAULT_VOICE
+        resolved_transcription = input_audio_transcription
+        resolved_turn_detection = turn_detection
+        if is_xai:
+            if not is_given(resolved_transcription):
+                resolved_transcription = _XAI_DEFAULT_INPUT_AUDIO_TRANSCRIPTION
+            if not is_given(resolved_turn_detection):
+                resolved_turn_detection = _XAI_DEFAULT_TURN_DETECTION
+
         super().__init__(
             model=model,
-            voice=voice,
+            voice=resolved_voice,
             modalities=modalities,
-            input_audio_transcription=input_audio_transcription,
+            input_audio_transcription=resolved_transcription,
             input_audio_noise_reduction=input_audio_noise_reduction,
-            turn_detection=turn_detection,
+            turn_detection=resolved_turn_detection,
             tool_choice=tool_choice,
             speed=speed,
             tracing=tracing,

tests/test_realtime/test_openai_inference_realtime_model.py

@@ -183,3 +183,44 @@ def test_new_api_does_not_expose_deprecated_temperature() -> None:
     parameters = inspect.signature(InferenceRealtimeModel).parameters
 
     assert "temperature" not in parameters
+
+
+async def test_xai_models_use_gateway_compatible_defaults(
+    paused_realtime_main: None,
+) -> None:
+    model = InferenceRealtimeModel(
+        "xai/grok-voice-latest",
+        api_key="key",
+        api_secret="secret",
+    )
+    session = model.session()
+
+    event = session._msg_ch.recv_nowait()
+    dumped = event.model_dump(exclude_unset=True) if hasattr(event, "model_dump") else event
+
+    assert model._opts.voice == "eve"
+    assert dumped["session"]["audio"]["input"]["transcription"]["model"] == "grok-transcribe"
+    assert dumped["session"]["audio"]["input"]["turn_detection"]["type"] == "server_vad"
+    await session.aclose()
+
+
+async def test_xai_gateway_defaults_can_be_overridden(
+    paused_realtime_main: None,
+) -> None:
+    model = InferenceRealtimeModel(
+        "xai/grok-voice-latest",
+        api_key="key",
+        api_secret="secret",
+        voice="Ara",
+        input_audio_transcription=None,
+        turn_detection=None,
+    )
+    session = model.session()
+
+    event = session._msg_ch.recv_nowait()
+    dumped = event.model_dump(exclude_unset=True) if hasattr(event, "model_dump") else event
+
+    assert model._opts.voice == "Ara"
+    assert dumped["session"]["audio"]["input"].get("transcription") is None
+    assert dumped["session"]["audio"]["input"].get("turn_detection") is None
+    await session.aclose()