diff --git a/examples/foundational/19-openai-realtime.py b/examples/foundational/19-openai-realtime.py index 0d59782f1..d10fbc129 100644 --- a/examples/foundational/19-openai-realtime.py +++ b/examples/foundational/19-openai-realtime.py @@ -37,7 +37,10 @@ from pipecat.services.openai.realtime.events import ( SemanticTurnDetection, SessionProperties, ) -from pipecat.services.openai.realtime.llm import OpenAIRealtimeLLMService +from pipecat.services.openai.realtime.llm import ( + OpenAIRealtimeLLMService, + OpenAIRealtimeLLMSettings, +) from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.daily.transport import DailyParams from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams @@ -137,22 +140,10 @@ transport_params = { async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): logger.info(f"Starting bot") - session_properties = SessionProperties( - audio=AudioConfiguration( - input=AudioInput( - transcription=InputAudioTranscription(), - # Set openai TurnDetection parameters. Not setting this at all will turn it - # on by default - turn_detection=SemanticTurnDetection(), - # Or set to False to disable openai turn detection and use transport VAD - # turn_detection=False, - noise_reduction=InputAudioNoiseReduction(type="near_field"), - ) - ), - # In this example we provide tools through the context, but you could - # alternatively provide them here. - # tools=tools, - instructions="""You are a helpful and friendly AI. + llm = OpenAIRealtimeLLMService( + api_key=os.getenv("OPENAI_API_KEY"), + settings=OpenAIRealtimeLLMSettings( + system_instruction="""You are a helpful and friendly AI. Act like a human, but remember that you aren't a human and that you can't do human things in the real world. Your voice and personality should be warm and engaging, with a lively and @@ -166,11 +157,23 @@ You are participating in a voice conversation. Keep your responses concise, shor unless specifically asked to elaborate on a topic. Remember, your responses should be short. Just one or two sentences, usually. Respond in English.""", - ) - - llm = OpenAIRealtimeLLMService( - api_key=os.getenv("OPENAI_API_KEY"), - session_properties=session_properties, + session_properties=SessionProperties( + audio=AudioConfiguration( + input=AudioInput( + transcription=InputAudioTranscription(), + # Set openai TurnDetection parameters. Not setting this at all will turn it + # on by default + turn_detection=SemanticTurnDetection(), + # Or set to False to disable openai turn detection and use transport VAD + # turn_detection=False, + noise_reduction=InputAudioNoiseReduction(type="near_field"), + ) + ), + # In this example we provide tools through the context, but you could + # alternatively provide them here. + # tools=tools, + ), + ), ) # you can either register a single function for all function calls, or specific functions diff --git a/examples/foundational/19a-azure-realtime.py b/examples/foundational/19a-azure-realtime.py index 6883a1f09..8d52a1c12 100644 --- a/examples/foundational/19a-azure-realtime.py +++ b/examples/foundational/19a-azure-realtime.py @@ -30,6 +30,7 @@ from pipecat.services.openai.realtime.events import ( InputAudioTranscription, SessionProperties, ) +from pipecat.services.openai.realtime.llm import OpenAIRealtimeLLMSettings from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.daily.transport import DailyParams from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams @@ -111,19 +112,11 @@ transport_params = { async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): logger.info(f"Starting bot") - session_properties = SessionProperties( - audio=AudioConfiguration( - input=AudioInput( - transcription=InputAudioTranscription(model="whisper-1"), - # Set openai TurnDetection parameters. Not setting this at all will turn it - # on by default - # turn_detection=TurnDetection(silence_duration_ms=1000), - # Or set to False to disable openai turn detection and use transport VAD - # turn_detection=False, - ) - ), - # tools=tools, - instructions="""You are a helpful and friendly AI. + llm = AzureRealtimeLLMService( + api_key=os.getenv("AZURE_REALTIME_API_KEY"), + base_url=os.getenv("AZURE_REALTIME_BASE_URL"), + settings=OpenAIRealtimeLLMSettings( + system_instruction="""You are a helpful and friendly AI. Act like a human, but remember that you aren't a human and that you can't do human things in the real world. Your voice and personality should be warm and engaging, with a lively and @@ -141,12 +134,20 @@ You have access to the following tools: - get_restaurant_recommendation: Get a restaurant recommendation for a given location. Remember, your responses should be short. Just one or two sentences, usually. Respond in English.""", - ) - - llm = AzureRealtimeLLMService( - api_key=os.getenv("AZURE_REALTIME_API_KEY"), - base_url=os.getenv("AZURE_REALTIME_BASE_URL"), - session_properties=session_properties, + session_properties=SessionProperties( + audio=AudioConfiguration( + input=AudioInput( + transcription=InputAudioTranscription(model="whisper-1"), + # Set openai TurnDetection parameters. Not setting this at all will turn it + # on by default + # turn_detection=TurnDetection(silence_duration_ms=1000), + # Or set to False to disable openai turn detection and use transport VAD + # turn_detection=False, + ) + ), + # tools=tools, + ), + ), ) # you can either register a single function for all function calls, or specific functions diff --git a/examples/foundational/19b-openai-realtime-text.py b/examples/foundational/19b-openai-realtime-text.py index 98dc81c3b..e03381642 100644 --- a/examples/foundational/19b-openai-realtime-text.py +++ b/examples/foundational/19b-openai-realtime-text.py @@ -32,7 +32,10 @@ from pipecat.services.openai.realtime.events import ( SemanticTurnDetection, SessionProperties, ) -from pipecat.services.openai.realtime.llm import OpenAIRealtimeLLMService +from pipecat.services.openai.realtime.llm import ( + OpenAIRealtimeLLMService, + OpenAIRealtimeLLMSettings, +) from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.daily.transport import DailyParams from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams @@ -113,21 +116,10 @@ transport_params = { async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): logger.info(f"Starting bot") - session_properties = SessionProperties( - audio=AudioConfiguration( - input=AudioInput( - transcription=InputAudioTranscription(), - # Set openai TurnDetection parameters. Not setting this at all will turn it - # on by default - turn_detection=SemanticTurnDetection(), - # Or set to False to disable openai turn detection and use transport VAD - # turn_detection=False, - noise_reduction=InputAudioNoiseReduction(type="near_field"), - ) - ), - output_modalities=["text"], - # tools=tools, - instructions="""You are a helpful and friendly AI. + llm = OpenAIRealtimeLLMService( + api_key=os.getenv("OPENAI_API_KEY"), + settings=OpenAIRealtimeLLMSettings( + system_instruction="""You are a helpful and friendly AI. Act like a human, but remember that you aren't a human and that you can't do human things in the real world. Your voice and personality should be warm and engaging, with a lively and @@ -145,11 +137,22 @@ You have access to the following tools: - get_restaurant_recommendation: Get a restaurant recommendation for a given location. Remember, your responses should be short. Just one or two sentences, usually. Respond in English.""", - ) - - llm = OpenAIRealtimeLLMService( - api_key=os.getenv("OPENAI_API_KEY"), - session_properties=session_properties, + session_properties=SessionProperties( + audio=AudioConfiguration( + input=AudioInput( + transcription=InputAudioTranscription(), + # Set openai TurnDetection parameters. Not setting this at all will turn it + # on by default + turn_detection=SemanticTurnDetection(), + # Or set to False to disable openai turn detection and use transport VAD + # turn_detection=False, + noise_reduction=InputAudioNoiseReduction(type="near_field"), + ) + ), + output_modalities=["text"], + # tools=tools, + ), + ), ) tts = CartesiaTTSService( diff --git a/examples/foundational/19c-openai-realtime-live-video.py b/examples/foundational/19c-openai-realtime-live-video.py index 3f091f712..31088ff0f 100644 --- a/examples/foundational/19c-openai-realtime-live-video.py +++ b/examples/foundational/19c-openai-realtime-live-video.py @@ -32,7 +32,10 @@ from pipecat.services.openai.realtime.events import ( SemanticTurnDetection, SessionProperties, ) -from pipecat.services.openai.realtime.llm import OpenAIRealtimeLLMService +from pipecat.services.openai.realtime.llm import ( + OpenAIRealtimeLLMService, + OpenAIRealtimeLLMSettings, +) from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.daily.transport import DailyParams @@ -60,22 +63,10 @@ transport_params = { async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): logger.info(f"Starting bot") - session_properties = SessionProperties( - audio=AudioConfiguration( - input=AudioInput( - transcription=InputAudioTranscription(), - # Set openai TurnDetection parameters. Not setting this at all will turn it - # on by default - turn_detection=SemanticTurnDetection(), - # Or set to False to disable openai turn detection and use transport VAD - # turn_detection=False, - noise_reduction=InputAudioNoiseReduction(type="near_field"), - ) - ), - # In this example we provide tools through the context, but you could - # alternatively provide them here. - # tools=tools, - instructions="""You are a helpful and friendly AI. + llm = OpenAIRealtimeLLMService( + api_key=os.getenv("OPENAI_API_KEY"), + settings=OpenAIRealtimeLLMSettings( + system_instruction="""You are a helpful and friendly AI. Act like a human, but remember that you aren't a human and that you can't do human things in the real world. Your voice and personality should be warm and engaging, with a lively and @@ -89,11 +80,23 @@ You are participating in a voice conversation. Keep your responses concise, shor unless specifically asked to elaborate on a topic. Remember, your responses should be short. Just one or two sentences, usually. Respond in English.""", - ) - - llm = OpenAIRealtimeLLMService( - api_key=os.getenv("OPENAI_API_KEY"), - session_properties=session_properties, + session_properties=SessionProperties( + audio=AudioConfiguration( + input=AudioInput( + transcription=InputAudioTranscription(), + # Set openai TurnDetection parameters. Not setting this at all will turn it + # on by default + turn_detection=SemanticTurnDetection(), + # Or set to False to disable openai turn detection and use transport VAD + # turn_detection=False, + noise_reduction=InputAudioNoiseReduction(type="near_field"), + ) + ), + # In this example we provide tools through the context, but you could + # alternatively provide them here. + # tools=tools, + ), + ), ) # Create a standard OpenAI LLM context object using the normal messages format. The diff --git a/examples/foundational/20b-persistent-context-openai-realtime.py b/examples/foundational/20b-persistent-context-openai-realtime.py index a24ff39e5..b85f8c319 100644 --- a/examples/foundational/20b-persistent-context-openai-realtime.py +++ b/examples/foundational/20b-persistent-context-openai-realtime.py @@ -33,7 +33,10 @@ from pipecat.services.openai.realtime.events import ( SessionProperties, TurnDetection, ) -from pipecat.services.openai.realtime.llm import OpenAIRealtimeLLMService +from pipecat.services.openai.realtime.llm import ( + OpenAIRealtimeLLMService, + OpenAIRealtimeLLMSettings, +) from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.daily.transport import DailyParams from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams @@ -173,19 +176,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) - session_properties = SessionProperties( - audio=AudioConfiguration( - input=AudioInput( - transcription=InputAudioTranscription(), - # Set openai TurnDetection parameters. Not setting this at all will turn it - # on by default - turn_detection=TurnDetection(silence_duration_ms=1000), - # Or set to False to disable openai turn detection and use transport VAD - # turn_detection=False, - ) - ), - # tools=tools, - instructions="""Your knowledge cutoff is 2023-10. You are a helpful and friendly AI. + llm = OpenAIRealtimeLLMService( + api_key=os.getenv("OPENAI_API_KEY"), + settings=OpenAIRealtimeLLMSettings( + system_instruction="""Your knowledge cutoff is 2023-10. You are a helpful and friendly AI. Act like a human, but remember that you aren't a human and that you can't do human things in the real world. Your voice and personality should be warm and engaging, with a lively and @@ -199,11 +193,20 @@ You are participating in a voice conversation. Keep your responses concise, shor unless specifically asked to elaborate on a topic. Remember, your responses should be short. Just one or two sentences, usually.""", - ) - - llm = OpenAIRealtimeLLMService( - api_key=os.getenv("OPENAI_API_KEY"), - session_properties=session_properties, + session_properties=SessionProperties( + audio=AudioConfiguration( + input=AudioInput( + transcription=InputAudioTranscription(), + # Set openai TurnDetection parameters. Not setting this at all will turn it + # on by default + turn_detection=TurnDetection(silence_duration_ms=1000), + # Or set to False to disable openai turn detection and use transport VAD + # turn_detection=False, + ) + ), + # tools=tools, + ), + ), ) # you can either register a single function for all function calls, or specific functions diff --git a/src/pipecat/services/openai/realtime/llm.py b/src/pipecat/services/openai/realtime/llm.py index b5d9dc8e2..36c6d8fae 100644 --- a/src/pipecat/services/openai/realtime/llm.py +++ b/src/pipecat/services/openai/realtime/llm.py @@ -10,8 +10,9 @@ import base64 import io import json import time -from dataclasses import dataclass -from typing import Any, Optional +from dataclasses import dataclass, field +from dataclasses import fields as dataclass_fields +from typing import Any, Dict, Mapping, Optional, Type from loguru import logger from PIL import Image @@ -59,7 +60,13 @@ from pipecat.processors.aggregators.openai_llm_context import ( ) from pipecat.processors.frame_processor import FrameDirection from pipecat.services.llm_service import FunctionCallFromLLM, LLMService -from pipecat.services.settings import LLMSettings, _warn_deprecated_param +from pipecat.services.settings import ( + NOT_GIVEN, + LLMSettings, + _NotGiven, + _warn_deprecated_param, + is_given, +) from pipecat.transcriptions.language import Language from pipecat.utils.time import time_now_iso8601 from pipecat.utils.tracing.service_decorators import traced_openai_realtime, traced_stt @@ -93,9 +100,107 @@ class CurrentAudioResponse: @dataclass class OpenAIRealtimeLLMSettings(LLMSettings): - """Settings for OpenAI Realtime LLM services.""" + """Settings for OpenAI Realtime LLM services. - pass + Parameters: + session_properties: OpenAI Realtime session properties (modalities, + audio config, tools, etc.). ``model`` and ``instructions`` are + synced bidirectionally with the top-level ``model`` and + ``system_instruction`` fields. + """ + + session_properties: events.SessionProperties | _NotGiven = field( + default_factory=lambda: NOT_GIVEN + ) + + # -- Bidirectional sync helpers ------------------------------------------ + + @staticmethod + def _sync_top_level_to_sp(settings: "OpenAIRealtimeLLMSettings"): + """Push top-level ``model``/``system_instruction`` into ``session_properties``.""" + if not is_given(settings.session_properties): + return + sp = settings.session_properties + if is_given(settings.model) and settings.model is not None: + sp.model = settings.model + if is_given(settings.system_instruction): + sp.instructions = settings.system_instruction + + # -- apply_update override ----------------------------------------------- + + def apply_update(self, delta: "OpenAIRealtimeLLMSettings") -> Dict[str, Any]: + """Merge a delta, keeping ``model``/``system_instruction`` in sync with SP. + + When the delta contains ``session_properties``, it **replaces** the + stored SP wholesale (matching legacy behaviour). Top-level field + values always take precedence over conflicting SP values. + """ + # 1. Let the base class handle all fields including session_properties + # (wholesale replacement when given). + changed = super().apply_update(delta) + + # 2. SP → top-level: if the SP was just replaced and carries + # model/instructions that the delta didn't set at top level, + # pull them up. + if "session_properties" in changed and is_given(self.session_properties): + sp = self.session_properties + if "model" not in changed and sp.model is not None: + old_model = self.model + self.model = sp.model + if old_model != self.model: + changed["model"] = old_model + if "system_instruction" not in changed and sp.instructions is not None: + old_si = self.system_instruction + self.system_instruction = sp.instructions + if old_si != self.system_instruction: + changed["system_instruction"] = old_si + + # 3. Top-level → SP: ensure SP mirrors the authoritative top-level + # values. Covers all cases: top-level-only delta, SP-only delta, + # and mixed deltas where top-level takes precedence. + self._sync_top_level_to_sp(self) + + return changed + + # -- from_mapping override ----------------------------------------------- + + @classmethod + def from_mapping( + cls: Type["OpenAIRealtimeLLMSettings"], settings: Mapping[str, Any] + ) -> "OpenAIRealtimeLLMSettings": + """Build a delta from a plain dict, routing SP keys into ``session_properties``. + + Keys that correspond to ``SessionProperties`` fields (except ``model``) + are collected into a nested ``session_properties`` value. ``model`` is + always routed to the top-level field. Unknown keys go to ``extra``. + """ + # Determine which keys belong to our own dataclass fields. + own_field_names = {f.name for f in dataclass_fields(cls)} - {"extra"} + + top: Dict[str, Any] = {} + sp_dict: Dict[str, Any] = {} + extra: Dict[str, Any] = {} + + # Build the SP field set without instantiating (avoid __post_init__ + # cost for every from_mapping call). + sp_keys = set(events.SessionProperties.model_fields.keys()) - {"model"} + + for key, value in settings.items(): + # Resolve aliases first + canonical = cls._aliases.get(key, key) + if canonical in own_field_names: + top[canonical] = value + elif canonical in sp_keys: + sp_dict[canonical] = value + else: + extra[key] = value + + if sp_dict: + top["session_properties"] = events.SessionProperties(**sp_dict) + + instance = cls(**top) + instance.extra = extra + return instance class OpenAIRealtimeLLMService(LLMService): @@ -140,6 +245,10 @@ class OpenAIRealtimeLLMService(LLMService): Defaults to "wss://api.openai.com/v1/realtime". session_properties: Configuration properties for the realtime session. If None, uses default SessionProperties. + + .. deprecated:: + Use ``settings=OpenAIRealtimeLLMSettings(session_properties=...)`` + instead. settings: Runtime-updatable settings for this service. start_audio_paused: Whether to start with audio input paused. Defaults to False. start_video_paused: Whether to start with video input paused. Defaults to False. @@ -180,6 +289,7 @@ class OpenAIRealtimeLLMService(LLMService): seed=None, filter_incomplete_user_turns=False, user_turn_completion_config=None, + session_properties=events.SessionProperties(), ) # 2. Apply direct init arg overrides (deprecated) @@ -187,6 +297,23 @@ class OpenAIRealtimeLLMService(LLMService): _warn_deprecated_param("model", OpenAIRealtimeLLMSettings, "model") default_settings.model = model + if session_properties is not None: + _warn_deprecated_param( + "session_properties", + OpenAIRealtimeLLMSettings, + "session_properties", + ) + default_settings.session_properties = session_properties + # Sync model/instructions from the deprecated SP arg to top-level, + # but only if the deprecated `model` arg didn't already set it. + if model is None and session_properties.model is not None: + default_settings.model = session_properties.model + if session_properties.instructions is not None: + default_settings.system_instruction = session_properties.instructions + + # Sync top-level model back into session_properties + OpenAIRealtimeLLMSettings._sync_top_level_to_sp(default_settings) + # 3. Apply settings delta (canonical API, always wins) if settings is not None: default_settings.apply_update(settings) @@ -202,7 +329,6 @@ class OpenAIRealtimeLLMService(LLMService): self.api_key = api_key self.base_url = full_url - self._session_properties = session_properties or events.SessionProperties() self._audio_input_paused = start_audio_paused self._video_input_paused = start_video_paused self._video_frame_detail = video_frame_detail @@ -265,12 +391,12 @@ class OpenAIRealtimeLLMService(LLMService): def _is_modality_enabled(self, modality: str) -> bool: """Check if a specific modality is enabled, "text" or "audio".""" - modalities = self._session_properties.output_modalities or ["audio", "text"] + modalities = self._settings.session_properties.output_modalities or ["audio", "text"] return modality in modalities def _get_enabled_modalities(self) -> list[str]: """Get the list of enabled modalities.""" - modalities = self._session_properties.output_modalities or ["audio", "text"] + modalities = self._settings.session_properties.output_modalities or ["audio", "text"] # API only supports single modality responses: either ["text"] or ["audio"] if "audio" in modalities: return ["audio"] @@ -343,9 +469,9 @@ class OpenAIRealtimeLLMService(LLMService): # None and False are different. Check for False. None means we're using OpenAI's # built-in turn detection defaults. turn_detection_disabled = ( - self._session_properties.audio - and self._session_properties.audio.input - and self._session_properties.audio.input.turn_detection is False + self._settings.session_properties.audio + and self._settings.session_properties.audio.input + and self._settings.session_properties.audio.input.turn_detection is False ) if turn_detection_disabled: await self.send_client_event(events.InputAudioBufferClearEvent()) @@ -365,9 +491,9 @@ class OpenAIRealtimeLLMService(LLMService): # None and False are different. Check for False. None means we're using OpenAI's # built-in turn detection defaults. turn_detection_disabled = ( - self._session_properties.audio - and self._session_properties.audio.input - and self._session_properties.audio.input.turn_detection is False + self._settings.session_properties.audio + and self._settings.session_properties.audio.input + and self._settings.session_properties.audio.input.turn_detection is False ) if turn_detection_disabled: await self.send_client_event(events.InputAudioBufferCommitEvent()) @@ -435,16 +561,6 @@ class OpenAIRealtimeLLMService(LLMService): frame: The frame to process. direction: The direction of frame flow in the pipeline. """ - # Backward-compatible dict path: frame.settings contains SessionProperties - # fields, not our Settings fields, so we construct SessionProperties - # directly. The frame.delta path falls through to super, which calls - # _update_settings → our override handles the rest. - if isinstance(frame, LLMUpdateSettingsFrame) and frame.delta is None: - self._session_properties = events.SessionProperties(**frame.settings) - await self._send_session_update() - await self.push_frame(frame, direction) - return - await super().process_frame(frame, direction) if isinstance(frame, TranscriptionFrame): @@ -559,13 +675,16 @@ class OpenAIRealtimeLLMService(LLMService): await self.push_error(error_msg=f"Error sending client event: {e}", exception=e) async def _update_settings(self, delta): - """Apply a settings delta.""" + """Apply a settings delta, sending a session update when needed.""" changed = await super()._update_settings(delta) - self._warn_unhandled_updated_settings(changed.keys()) + handled = {"session_properties", "system_instruction"} + if changed.keys() & handled: + await self._send_session_update() + self._warn_unhandled_updated_settings(changed.keys() - handled) return changed async def _send_session_update(self): - settings = self._session_properties + settings = self._settings.session_properties adapter: OpenAIRealtimeLLMAdapter = self.get_llm_adapter() if self._context: diff --git a/tests/test_settings.py b/tests/test_settings.py index 201e85745..875ccf067 100644 --- a/tests/test_settings.py +++ b/tests/test_settings.py @@ -12,6 +12,8 @@ import pytest from pipecat.services.deepgram.stt import DeepgramSTTService, DeepgramSTTSettings from pipecat.services.deepgram.stt_sagemaker import DeepgramSageMakerSTTSettings +from pipecat.services.openai.realtime import events +from pipecat.services.openai.realtime.llm import OpenAIRealtimeLLMSettings from pipecat.services.settings import ( NOT_GIVEN, LLMSettings, @@ -615,3 +617,201 @@ class TestDeepgramSTTSettingsExtraSync: kwargs = svc._build_connect_kwargs() assert kwargs["numerals"] == "true" assert kwargs["custom_param"] == "test" + + +# --------------------------------------------------------------------------- +# OpenAIRealtimeLLMSettings: apply_update with bidirectional sync +# --------------------------------------------------------------------------- + + +class TestOpenAIRealtimeSettingsApplyUpdate: + def _make_store(self, **kwargs) -> OpenAIRealtimeLLMSettings: + """Helper to build a store-mode OpenAIRealtimeLLMSettings.""" + defaults = dict( + model="gpt-realtime-1.5", + system_instruction=None, + temperature=None, + max_tokens=None, + top_p=None, + top_k=None, + frequency_penalty=None, + presence_penalty=None, + seed=None, + filter_incomplete_user_turns=False, + user_turn_completion_config=None, + session_properties=events.SessionProperties(), + ) + defaults.update(kwargs) + return OpenAIRealtimeLLMSettings(**defaults) + + def test_top_level_model_syncs_to_sp(self): + """Updating top-level model should propagate to session_properties.model.""" + store = self._make_store() + delta = OpenAIRealtimeLLMSettings(model="gpt-realtime-2.0") + changed = store.apply_update(delta) + + assert "model" in changed + assert store.model == "gpt-realtime-2.0" + assert store.session_properties.model == "gpt-realtime-2.0" + + def test_top_level_system_instruction_syncs_to_sp(self): + """Updating top-level system_instruction should propagate to session_properties.instructions.""" + store = self._make_store() + delta = OpenAIRealtimeLLMSettings(system_instruction="Be helpful.") + changed = store.apply_update(delta) + + assert "system_instruction" in changed + assert store.system_instruction == "Be helpful." + assert store.session_properties.instructions == "Be helpful." + + def test_sp_replaces_wholesale(self): + """session_properties in delta replaces the entire stored SP.""" + store = self._make_store( + session_properties=events.SessionProperties( + output_modalities=["audio", "text"], + instructions="Old instructions.", + ), + system_instruction="Old instructions.", + ) + + new_sp = events.SessionProperties(output_modalities=["text"]) + delta = OpenAIRealtimeLLMSettings(session_properties=new_sp) + changed = store.apply_update(delta) + + assert "session_properties" in changed + assert store.session_properties.output_modalities == ["text"] + # Fields not in the new SP become None (wholesale replacement) + # But model is synced from top-level + assert store.session_properties.model == "gpt-realtime-1.5" + + def test_sp_model_syncs_to_top_level(self): + """session_properties.model should sync to top-level model.""" + store = self._make_store() + new_sp = events.SessionProperties(model="gpt-realtime-2.0") + delta = OpenAIRealtimeLLMSettings(session_properties=new_sp) + changed = store.apply_update(delta) + + assert "model" in changed + assert store.model == "gpt-realtime-2.0" + assert store.session_properties.model == "gpt-realtime-2.0" + + def test_sp_instructions_syncs_to_top_level(self): + """session_properties.instructions should sync to top-level system_instruction.""" + store = self._make_store() + new_sp = events.SessionProperties(instructions="New instructions.") + delta = OpenAIRealtimeLLMSettings(session_properties=new_sp) + changed = store.apply_update(delta) + + assert "system_instruction" in changed + assert store.system_instruction == "New instructions." + assert store.session_properties.instructions == "New instructions." + + def test_top_level_model_takes_precedence_over_sp_model(self): + """When both model and session_properties.model are in the delta, top-level wins.""" + store = self._make_store() + new_sp = events.SessionProperties(model="sp-model") + delta = OpenAIRealtimeLLMSettings(model="top-model", session_properties=new_sp) + store.apply_update(delta) + + assert store.model == "top-model" + assert store.session_properties.model == "top-model" + + def test_top_level_si_takes_precedence_over_sp_instructions(self): + """When both system_instruction and SP.instructions are in delta, top-level wins.""" + store = self._make_store() + new_sp = events.SessionProperties(instructions="sp instructions") + delta = OpenAIRealtimeLLMSettings( + system_instruction="top instructions", + session_properties=new_sp, + ) + store.apply_update(delta) + + assert store.system_instruction == "top instructions" + assert store.session_properties.instructions == "top instructions" + + def test_non_synced_field_update_does_not_affect_sp(self): + """Updating a non-synced field like temperature shouldn't touch session_properties.""" + store = self._make_store( + session_properties=events.SessionProperties(instructions="Keep me."), + system_instruction="Keep me.", + ) + original_sp = store.session_properties + + delta = OpenAIRealtimeLLMSettings(temperature=0.5) + changed = store.apply_update(delta) + + assert "temperature" in changed + assert store.temperature == 0.5 + # SP should be untouched (same object) + assert store.session_properties is original_sp + assert store.session_properties.instructions == "Keep me." + + +# --------------------------------------------------------------------------- +# OpenAIRealtimeLLMSettings: from_mapping +# --------------------------------------------------------------------------- + + +class TestOpenAIRealtimeSettingsFromMapping: + def test_sp_keys_route_to_session_properties(self): + """SessionProperties fields (instructions, audio, etc.) route into nested SP.""" + delta = OpenAIRealtimeLLMSettings.from_mapping( + {"instructions": "Be concise.", "output_modalities": ["text"]} + ) + assert is_given(delta.session_properties) + assert delta.session_properties.instructions == "Be concise." + assert delta.session_properties.output_modalities == ["text"] + + def test_model_routes_to_top_level(self): + """model should go to the top-level field, not session_properties.""" + delta = OpenAIRealtimeLLMSettings.from_mapping({"model": "gpt-realtime-2.0"}) + assert delta.model == "gpt-realtime-2.0" + # No session_properties should be created since no SP keys were present + assert not is_given(delta.session_properties) + + def test_unknown_keys_go_to_extra(self): + """Unrecognized keys should land in extra.""" + delta = OpenAIRealtimeLLMSettings.from_mapping({"unknown_param": 42}) + assert not is_given(delta.model) + assert not is_given(delta.session_properties) + assert delta.extra == {"unknown_param": 42} + + def test_mixed_keys(self): + """model + SP keys + unknown keys are routed correctly.""" + delta = OpenAIRealtimeLLMSettings.from_mapping( + { + "model": "gpt-realtime-2.0", + "instructions": "Be helpful.", + "unknown": "val", + } + ) + assert delta.model == "gpt-realtime-2.0" + assert is_given(delta.session_properties) + assert delta.session_properties.instructions == "Be helpful." + assert delta.extra == {"unknown": "val"} + + def test_roundtrip_from_mapping_apply_update(self): + """Simulate dict-style update: from_mapping -> apply_update.""" + store = OpenAIRealtimeLLMSettings( + model="gpt-realtime-1.5", + system_instruction=None, + temperature=None, + max_tokens=None, + top_p=None, + top_k=None, + frequency_penalty=None, + presence_penalty=None, + seed=None, + filter_incomplete_user_turns=False, + user_turn_completion_config=None, + session_properties=events.SessionProperties(), + ) + + raw = {"instructions": "Be concise.", "output_modalities": ["text"]} + delta = OpenAIRealtimeLLMSettings.from_mapping(raw) + changed = store.apply_update(delta) + + assert "session_properties" in changed + assert store.session_properties.instructions == "Be concise." + assert store.session_properties.output_modalities == ["text"] + assert store.system_instruction == "Be concise."