wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #3939 from pipecat-ai/pk/openai-realtime-settings-pattern

Browse Source 
Adopt the `settings` pattern for OpenAI Realtime session properties
This commit is contained in:
kompfner
2026-03-06 12:39:08 -05:00
committed by GitHub
parent 18494658c3 2b8a6d9ca4
commit d84a250b62
7 changed files with 462 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
examples/foundational/19-openai-realtime.py
View File

@@ -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

39
examples/foundational/19a-azure-realtime.py
View File

@@ -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

45
examples/foundational/19b-openai-realtime-text.py
View File

@@ -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(

47
examples/foundational/19c-openai-realtime-live-video.py
View File

@@ -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

41
examples/foundational/20b-persistent-context-openai-realtime.py
View File

@@ -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

173
src/pipecat/services/openai/realtime/llm.py
View File

@@ -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:

200
tests/test_settings.py
View File

@@ -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."