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

Use ServiceSettings types rather than dicts to represent changed—the return value of ServiceSettings.apply_update—allowing users to avoid stringly-typed code in their _update_settings overrides.

Browse Source
This commit is contained in:
Paul Kompfner
2026-03-11 15:10:44 -04:00
parent 65561a1d83
commit eedd89186d
43 changed files with 261 additions and 231 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
COMMUNITY_INTEGRATIONS.md
View File

@@ -325,14 +325,14 @@ svc = MyTTSService(
AI services support runtime configuration changes via `*UpdateSettingsFrame`s (e.g. `STTUpdateSettingsFrame`, `TTSUpdateSettingsFrame`, `LLMUpdateSettingsFrame`).
To react to runtime setting changes, override `_update_settings`. The base implementation applies the delta to `self._settings` and returns a `dict` mapping each changed field name to its **pre-update** value. Your override should call `super()` first, then act on the changed fields. A common implementation might look like:
To react to runtime setting changes, override `_update_settings`. The base implementation applies the delta to `self._settings` and returns a delta-mode settings object of the same type with changed fields set to their **pre-update** values (unchanged fields are `NOT_GIVEN`). Your override should call `super()` first, then act on the changed fields. A common implementation might look like:
```python
async def _update_settings(self, update: TTSSettings) -> dict[str, Any]:
async def _update_settings(self, update: TTSSettings) -> TTSSettings:
"""Apply a settings update, reconfiguring the connection if needed."""
changed = await super()._update_settings(update)
if not changed:
if not changed.given_fields():
return changed
await self._disconnect()
@@ -341,24 +341,24 @@ async def _update_settings(self, update: TTSSettings) -> dict[str, Any]:
return changed
```
The dict keys work like a set for membership tests (`"language" in changed`) and truthiness (`if changed`). Use `changed.keys() - {"language"}` for set difference, or `changed["language"]` to inspect the previous value of a field.
Use `is_given(changed.language)` to check whether a specific field changed, and `changed.given_fields()` to get a dict of all changed fields. Use `changed.given_fields().keys() - {"language"}` for set difference, or access old values directly with typed attributes (e.g. `changed.language`).
Note that, in this example, the service requires a reconnect to apply the new language. Consider, for each setting, whether your service requires reconnection or can apply changes in-place.
If your service can't yet apply certain settings at runtime, call `self._warn_unhandled_updated_settings(changed)` with any unhandled field names so users get a clear log message:
If your service can't yet apply certain settings at runtime, call `self._warn_unhandled_updated_settings(changed.given_fields())` with any unhandled field names so users get a clear log message:
```python
async def _update_settings(self, update: TTSSettings) -> dict[str, Any]:
async def _update_settings(self, update: TTSSettings) -> TTSSettings:
changed = await super()._update_settings(update)
if not changed:
if not changed.given_fields():
return changed
if "language" in changed:
if is_given(changed.language):
await self._update_language()
else:
# TODO: this should be temporary - handle changes to other settings soon!
self._warn_unhandled_updated_settings(changed.keys() - {"language"})
self._warn_unhandled_updated_settings(changed.given_fields().keys() - {"language"})
return changed
```

22
src/pipecat/services/ai_service.py
View File

@@ -24,7 +24,7 @@ from pipecat.frames.frames import (
)
from pipecat.metrics.metrics import MetricsData
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.services.settings import ServiceSettings
from pipecat.services.settings import ServiceSettings, is_given
class AIService(FrameProcessor):
@@ -104,13 +104,13 @@ class AIService(FrameProcessor):
"""
pass
async def _update_settings(self, delta: ServiceSettings) -> Dict[str, Any]:
async def _update_settings(self, delta: ServiceSettings) -> ServiceSettings:
"""Apply a settings delta and return the changed fields.
The delta is applied to ``_settings`` and a dict mapping each changed
field name to its **pre-update** value is returned. The ``model``
field is handled specially: when it changes, ``set_model_name`` is
called.
The delta is applied to ``_settings`` and a delta-mode settings object
is returned with changed fields set to their **pre-update** values.
The ``model`` field is handled specially: when it changes,
``set_model_name`` is called.
Concrete services should override this method (calling ``super()``)
to react to specific changed fields (e.g. reconnect on voice change).
@@ -119,15 +119,17 @@ class AIService(FrameProcessor):
delta: A delta-mode settings object.
Returns:
Dict mapping changed field names to their previous values.
A delta-mode settings object with changed fields set to their
previous values. Unchanged fields are ``NOT_GIVEN``.
"""
changed = self._settings.apply_update(delta)
if "model" in changed:
if is_given(changed.model):
self._sync_model_name_to_metrics()
if changed:
logger.info(f"{self.name}: updated settings fields: {set(changed)}")
changed_fields = changed.given_fields()
if changed_fields:
logger.info(f"{self.name}: updated settings fields: {set(changed_fields)}")
return changed

6
src/pipecat/services/assemblyai/stt.py
View File

@@ -353,18 +353,18 @@ class AssemblyAISTTService(WebsocketSTTService):
"""
return True
async def _update_settings(self, delta: Settings) -> dict[str, Any]:
async def _update_settings(self, delta: Settings) -> STTSettings:
"""Apply a settings delta and reconnect to apply changes.
Args:
delta: A settings delta with updated values.
Returns:
Dict mapping changed field names to their previous values.
A delta-mode settings object with changed fields.
"""
changed = await super()._update_settings(delta)
if not changed:
if not changed.given_fields():
return changed
# Reconnect to apply updated settings (they become WS query params)

6
src/pipecat/services/asyncai/tts.py
View File

@@ -202,17 +202,17 @@ class AsyncAITTSService(WebsocketTTSService):
self._receive_task = None
self._keepalive_task = None
async def _update_settings(self, delta: TTSSettings) -> dict[str, Any]:
async def _update_settings(self, delta: TTSSettings) -> Settings:
"""Apply a settings delta.
Settings are stored but not applied to the active connection.
"""
changed = await super()._update_settings(delta)
if not changed:
if not changed.given_fields():
return changed
self._warn_unhandled_updated_settings(changed)
self._warn_unhandled_updated_settings(changed.given_fields())
return changed

6
src/pipecat/services/aws/nova_sonic/llm.py
View File

@@ -445,14 +445,14 @@ class AWSNovaSonicLLMService(LLMService):
# settings
#
async def _update_settings(self, delta: Settings) -> dict[str, Any]:
async def _update_settings(self, delta: Settings) -> Settings:
"""Apply a settings delta.
Settings are stored but not applied to the active connection.
"""
changed = await super()._update_settings(delta)
if not changed:
if not changed.given_fields():
return changed
# TODO: someday we could reconnect here to apply updated settings.
@@ -460,7 +460,7 @@ class AWSNovaSonicLLMService(LLMService):
# await self._disconnect()
# await self._start_connecting()
self._warn_unhandled_updated_settings(changed)
self._warn_unhandled_updated_settings(changed.given_fields())
return changed

6
src/pipecat/services/aws/stt.py
View File

@@ -15,7 +15,7 @@ import os
import random
import string
from dataclasses import dataclass
from typing import Any, AsyncGenerator, Optional
from typing import AsyncGenerator, Optional
from loguru import logger
@@ -158,11 +158,11 @@ class AWSTranscribeSTTService(WebsocketSTTService):
}
return encoding_map.get(encoding, encoding)
async def _update_settings(self, delta: STTSettings) -> dict[str, Any]:
async def _update_settings(self, delta: STTSettings) -> Settings:
"""Apply a settings delta and reconnect if anything changed."""
changed = await super()._update_settings(delta)
if changed and self._websocket:
if changed.given_fields() and self._websocket:
await self._disconnect()
await self._connect()

8
src/pipecat/services/azure/stt.py
View File

@@ -12,7 +12,7 @@ Speech SDK for real-time audio transcription.
import asyncio
from dataclasses import dataclass
from typing import Any, AsyncGenerator, Optional
from typing import AsyncGenerator, Optional
from loguru import logger
@@ -26,7 +26,7 @@ from pipecat.frames.frames import (
TranscriptionFrame,
)
from pipecat.services.azure.common import language_to_azure_language
from pipecat.services.settings import STTSettings
from pipecat.services.settings import STTSettings, is_given
from pipecat.services.stt_latency import AZURE_TTFS_P99
from pipecat.services.stt_service import STTService
from pipecat.transcriptions.language import Language
@@ -176,11 +176,11 @@ class AzureSTTService(STTService):
"""
return language_to_azure_language(language)
async def _update_settings(self, delta: STTSettings) -> dict[str, Any]:
async def _update_settings(self, delta: STTSettings) -> Settings:
"""Apply a settings delta and reconnect if language changed."""
changed = await super()._update_settings(delta)
if "language" in changed:
if is_given(changed.language):
self._speech_config.speech_recognition_language = (
self._settings.language or language_to_azure_language(Language.EN_US)
)

11
src/pipecat/services/cartesia/stt.py
View File

@@ -13,7 +13,7 @@ the Cartesia Live transcription API for real-time speech recognition.
import json
import urllib.parse
from dataclasses import dataclass
from typing import Any, AsyncGenerator, Optional
from typing import AsyncGenerator, Optional
from loguru import logger
@@ -309,24 +309,25 @@ class CartesiaSTTService(WebsocketSTTService):
await self._disconnect_websocket()
async def _update_settings(self, delta: STTSettings) -> dict[str, Any]:
async def _update_settings(self, delta: STTSettings) -> Settings:
"""Apply a settings delta.
Args:
delta: A :class:`STTSettings` (or ``CartesiaSTTService.Settings``) delta.
Returns:
Dict mapping changed field names to their previous values.
A delta-mode settings object whose ``given_fields()`` contains
only the fields that actually changed.
"""
changed = await super()._update_settings(delta)
# TODO: someday we could reconnect here to apply updated settings.
# Code might look something like the below:
# if changed:
# if changed.given_fields():
# await self._disconnect()
# await self._connect()
self._warn_unhandled_updated_settings(changed)
self._warn_unhandled_updated_settings(changed.given_fields())
return changed

12
src/pipecat/services/deepgram/flux/stt.py
View File

@@ -28,7 +28,7 @@ from pipecat.frames.frames import (
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
)
from pipecat.services.settings import NOT_GIVEN, STTSettings, _NotGiven
from pipecat.services.settings import NOT_GIVEN, STTSettings, _NotGiven, is_given
from pipecat.services.stt_service import WebsocketSTTService
from pipecat.transcriptions.language import Language
from pipecat.utils.time import time_now_iso8601
@@ -448,7 +448,7 @@ class DeepgramFluxSTTService(WebsocketSTTService):
"""
return True
async def _update_settings(self, delta: Settings) -> dict[str, Any]:
async def _update_settings(self, delta: Settings) -> STTSettings:
"""Apply a settings delta.
Configure-able fields (keyterm, eot_threshold, eager_eot_threshold,
@@ -457,14 +457,16 @@ class DeepgramFluxSTTService(WebsocketSTTService):
"""
changed = await super()._update_settings(delta)
if not changed:
if not changed.given_fields():
return changed
configure_fields = changed.keys() & self._CONFIGURE_FIELDS
configure_fields = changed.given_fields().keys() & self._CONFIGURE_FIELDS
if configure_fields and self._websocket and self._websocket.state is State.OPEN:
await self._send_configure(configure_fields)
self._warn_unhandled_updated_settings(changed.keys() - self._CONFIGURE_FIELDS)
self._warn_unhandled_updated_settings(
changed.given_fields().keys() - self._CONFIGURE_FIELDS
)
return changed

6
src/pipecat/services/deepgram/sagemaker/stt.py
View File

@@ -208,11 +208,11 @@ class DeepgramSageMakerSTTService(STTService):
"""
return True
async def _update_settings(self, delta: STTSettings) -> dict[str, Any]:
async def _update_settings(self, delta: STTSettings) -> STTSettings:
"""Apply a settings delta and warn about unhandled changes."""
changed = await super()._update_settings(delta)
if not changed:
if not changed.given_fields():
return changed
# Sync extra to fields after the update so self._settings stays unambiguous
@@ -224,7 +224,7 @@ class DeepgramSageMakerSTTService(STTService):
# await self._disconnect()
# await self._connect()
self._warn_unhandled_updated_settings(changed)
self._warn_unhandled_updated_settings(changed.given_fields())
return changed

12
src/pipecat/services/deepgram/sagemaker/tts.py
View File

@@ -15,7 +15,7 @@ streaming audio output.
import asyncio
import json
from dataclasses import dataclass
from typing import Any, AsyncGenerator, Optional
from typing import AsyncGenerator, Optional
from loguru import logger
@@ -33,7 +33,7 @@ from pipecat.frames.frames import (
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.aws.sagemaker.bidi_client import SageMakerBidiClient
from pipecat.services.settings import TTSSettings
from pipecat.services.settings import TTSSettings, is_given
from pipecat.services.tts_service import TTSService
from pipecat.utils.tracing.service_decorators import traced_tts
@@ -234,7 +234,7 @@ class DeepgramSageMakerTTSService(TTSService):
logger.debug("Disconnected from Deepgram TTS on SageMaker")
await self._call_event_handler("on_disconnected")
async def _update_settings(self, delta: TTSSettings) -> dict[str, Any]:
async def _update_settings(self, delta: TTSSettings) -> TTSSettings:
"""Apply a settings delta and reconnect if necessary.
Since all settings are part of the SageMaker session query string,
@@ -242,11 +242,11 @@ class DeepgramSageMakerTTSService(TTSService):
"""
changed = await super()._update_settings(delta)
if not changed:
if not changed.given_fields():
return changed
# Deepgram uses voice as the model, so keep them in sync for metrics
if "voice" in changed:
if is_given(changed.voice):
self._settings.model = self._settings.voice
self._sync_model_name_to_metrics()
@@ -255,7 +255,7 @@ class DeepgramSageMakerTTSService(TTSService):
# await self._disconnect()
# await self._connect()
self._warn_unhandled_updated_settings(changed)
self._warn_unhandled_updated_settings(changed.given_fields())
return changed

4
src/pipecat/services/deepgram/stt.py
View File

@@ -485,11 +485,11 @@ class DeepgramSTTService(STTService):
"""
return True
async def _update_settings(self, delta: STTSettings) -> dict[str, Any]:
async def _update_settings(self, delta: STTSettings) -> STTSettings:
"""Apply a settings delta and reconnect if anything changed."""
changed = await super()._update_settings(delta)
if not changed:
if not changed.given_fields():
return changed
# Sync extra to fields after the update so self._settings stays unambiguous

12
src/pipecat/services/deepgram/tts.py
View File

@@ -12,7 +12,7 @@ for generating speech from text using various voice models.
import json
from dataclasses import dataclass
from typing import Any, AsyncGenerator, Optional
from typing import AsyncGenerator, Optional
import aiohttp
from loguru import logger
@@ -26,7 +26,7 @@ from pipecat.frames.frames import (
TTSAudioRawFrame,
TTSStoppedFrame,
)
from pipecat.services.settings import TTSSettings
from pipecat.services.settings import TTSSettings, is_given
from pipecat.services.tts_service import TTSService, WebsocketTTSService
from pipecat.utils.tracing.service_decorators import traced_tts
@@ -185,23 +185,23 @@ class DeepgramTTSService(WebsocketTTSService):
await self._disconnect_websocket()
async def _update_settings(self, delta: TTSSettings) -> dict[str, Any]:
async def _update_settings(self, delta: TTSSettings) -> TTSSettings:
"""Apply a settings delta.
Args:
delta: A :class:`TTSSettings` (or ``DeepgramTTSService.Settings``) delta.
Returns:
Dict mapping changed field names to their previous values.
A delta-mode settings object with changed fields.
"""
changed = await super()._update_settings(delta)
# Deepgram uses voice as the model, so keep them in sync for metrics
if "voice" in changed:
if is_given(changed.voice):
self._settings.model = self._settings.voice
self._sync_model_name_to_metrics()
if changed:
if changed.given_fields():
await self._disconnect()
await self._connect()

9
src/pipecat/services/elevenlabs/stt.py
View File

@@ -17,7 +17,7 @@ import io
import json
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, AsyncGenerator, Optional
from typing import AsyncGenerator, Optional
import aiohttp
from loguru import logger
@@ -593,18 +593,19 @@ class ElevenLabsRealtimeSTTService(WebsocketSTTService):
"""
return True
async def _update_settings(self, delta: STTSettings) -> dict[str, Any]:
async def _update_settings(self, delta: STTSettings) -> Settings:
"""Apply a settings delta and reconnect if anything changed.
Args:
delta: A :class:`STTSettings` (or ``ElevenLabsRealtimeSTTService.Settings``) delta.
Returns:
Dict mapping changed field names to their previous values.
A delta-mode settings object whose ``given_fields()`` contains
only the fields that actually changed.
"""
changed = await super()._update_settings(delta)
if not changed:
if not changed.given_fields():
return changed
if self._websocket:

26
src/pipecat/services/elevenlabs/tts.py
View File

@@ -529,7 +529,7 @@ class ElevenLabsTTSService(WebsocketTTSService):
def _set_voice_settings(self):
return build_elevenlabs_voice_settings(self._settings)
async def _update_settings(self, delta: TTSSettings) -> dict[str, Any]:
async def _update_settings(self, delta: TTSSettings) -> Settings:
"""Apply a settings delta, reconnecting as needed.
Uses the declarative ``URL_FIELDS`` and ``VOICE_SETTINGS_FIELDS``
@@ -540,29 +540,32 @@ class ElevenLabsTTSService(WebsocketTTSService):
delta: A :class:`TTSSettings` (or ``ElevenLabsTTSService.Settings``) delta.
Returns:
Dict mapping changed field names to their previous values.
A delta-mode settings object whose ``given_fields()`` contains
only the fields that actually changed.
"""
changed = await super()._update_settings(delta)
if not changed:
if not changed.given_fields():
return changed
# Rebuild voice settings for next context
self._voice_settings = self._set_voice_settings()
url_changed = bool(changed.keys() & self.Settings.URL_FIELDS)
voice_settings_changed = bool(changed.keys() & self.Settings.VOICE_SETTINGS_FIELDS)
url_changed = bool(changed.given_fields().keys() & self.Settings.URL_FIELDS)
voice_settings_changed = bool(
changed.given_fields().keys() & self.Settings.VOICE_SETTINGS_FIELDS
)
if url_changed:
logger.debug(
f"URL-level setting changed ({changed.keys() & self.Settings.URL_FIELDS}), "
f"URL-level setting changed ({changed.given_fields().keys() & self.Settings.URL_FIELDS}), "
f"reconnecting WebSocket"
)
await self._disconnect()
await self._connect()
elif voice_settings_changed:
logger.debug(
f"Voice settings changed ({changed.keys() & self.Settings.VOICE_SETTINGS_FIELDS}), "
f"Voice settings changed ({changed.given_fields().keys() & self.Settings.VOICE_SETTINGS_FIELDS}), "
f"closing current context to apply changes"
)
audio_contexts = self.get_audio_contexts()
@@ -574,7 +577,7 @@ class ElevenLabsTTSService(WebsocketTTSService):
# Reconnect applies all settings; only warn about fields not handled
# by voice settings or URL changes.
handled = self.Settings.URL_FIELDS | self.Settings.VOICE_SETTINGS_FIELDS
self._warn_unhandled_updated_settings(changed.keys() - handled)
self._warn_unhandled_updated_settings(changed.given_fields().keys() - handled)
return changed
@@ -1087,17 +1090,18 @@ class ElevenLabsHttpTTSService(TTSService):
def _set_voice_settings(self):
return build_elevenlabs_voice_settings(self._settings)
async def _update_settings(self, delta: TTSSettings) -> dict[str, Any]:
async def _update_settings(self, delta: TTSSettings) -> Settings:
"""Apply a settings delta and rebuild voice settings.
Args:
delta: A :class:`TTSSettings` (or ``ElevenLabsHttpTTSService.Settings``) delta.
Returns:
Dict mapping changed field names to their previous values.
A delta-mode settings object whose ``given_fields()`` contains
only the fields that actually changed.
"""
changed = await super()._update_settings(delta)
if changed:
if changed.given_fields():
self._voice_settings = self._set_voice_settings()
return changed

6
src/pipecat/services/fish/tts.py
View File

@@ -234,7 +234,7 @@ class FishAudioTTSService(InterruptibleTTSService):
"""
return True
async def _update_settings(self, delta: TTSSettings) -> dict[str, Any]:
async def _update_settings(self, delta: TTSSettings) -> Settings:
"""Apply a settings delta and reconnect if needed.
Any change to voice or model triggers a WebSocket reconnect.
@@ -243,11 +243,11 @@ class FishAudioTTSService(InterruptibleTTSService):
delta: A :class:`TTSSettings` (or ``FishAudioTTSService.Settings``) delta.
Returns:
Dict mapping changed field names to their previous values.
A delta-mode settings object with only the changed fields set.
"""
changed = await super()._update_settings(delta)
if changed:
if changed.given_fields():
await self._disconnect()
await self._connect()

8
src/pipecat/services/gladia/stt.py
View File

@@ -469,7 +469,7 @@ class GladiaSTTService(WebsocketSTTService):
await super().start(frame)
await self._connect()
async def _update_settings(self, delta: Settings) -> dict[str, Any]:
async def _update_settings(self, delta: Settings) -> Settings:
"""Apply settings delta.
Settings are stored but not applied to the active session.
@@ -478,11 +478,11 @@ class GladiaSTTService(WebsocketSTTService):
delta: A settings delta.
Returns:
Dict mapping changed field names to their previous values.
A delta-mode settings object with only the changed fields set.
"""
changed = await super()._update_settings(delta)
if not changed:
if not changed.given_fields():
return changed
# TODO: someday we could reconnect here to apply updated settings.
@@ -492,7 +492,7 @@ class GladiaSTTService(WebsocketSTTService):
# await self._disconnect()
# await self._connect()
self._warn_unhandled_updated_settings(changed)
self._warn_unhandled_updated_settings(changed.given_fields())
return changed

6
src/pipecat/services/google/gemini_live/llm.py
View File

@@ -870,14 +870,14 @@ class GeminiLiveLLMService(LLMService):
"""
return True
async def _update_settings(self, delta: LLMSettings) -> dict[str, Any]:
async def _update_settings(self, delta: LLMSettings) -> LLMSettings:
"""Apply a settings delta.
Settings are stored but not applied to the active connection.
"""
changed = await super()._update_settings(delta)
if not changed:
if not changed.given_fields():
return changed
# TODO: someday we could reconnect here to apply updated settings.
@@ -885,7 +885,7 @@ class GeminiLiveLLMService(LLMService):
# await self._disconnect()
# await self._connect()
self._warn_unhandled_updated_settings(changed)
self._warn_unhandled_updated_settings(changed.given_fields())
return changed

7
src/pipecat/services/google/stt.py
View File

@@ -671,7 +671,7 @@ class GoogleSTTService(STTService):
logger.debug(f"Switching STT languages to: {languages}")
await self._update_settings(self.Settings(languages=list(languages)))
async def _update_settings(self, delta: Settings) -> dict[str, Any]:
async def _update_settings(self, delta: Settings) -> Settings:
"""Apply settings delta and reconnect if anything changed.
Handles ``language`` from base ``set_language`` by converting it to
@@ -683,7 +683,8 @@ class GoogleSTTService(STTService):
delta: A settings delta.
Returns:
Dict mapping changed field names to their previous values.
A delta-mode settings object whose given fields are the ones that
actually changed (values are the *previous* values).
"""
from pipecat.services.settings import is_given
@@ -706,7 +707,7 @@ class GoogleSTTService(STTService):
changed = await super()._update_settings(delta)
if changed:
if changed.given_fields():
await self._reconnect_if_needed()
return changed

9
src/pipecat/services/google/tts.py
View File

@@ -740,7 +740,7 @@ class GoogleHttpTTSService(TTSService):
"""
return language_to_google_tts_language(language)
async def _update_settings(self, delta: TTSSettings) -> dict[str, Any]:
async def _update_settings(self, delta: TTSSettings) -> TTSSettings:
"""Override to handle speaking_rate validation.
Args:
@@ -1112,7 +1112,7 @@ class GoogleTTSService(GoogleBaseTTSService):
credentials, credentials_path
)
async def _update_settings(self, delta: TTSSettings) -> dict[str, Any]:
async def _update_settings(self, delta: TTSSettings) -> TTSSettings:
"""Override to handle speaking_rate validation.
Args:
@@ -1395,14 +1395,15 @@ class GeminiTTSService(GoogleBaseTTSService):
f"Current rate of {self.sample_rate}Hz may cause issues."
)
async def _update_settings(self, delta: TTSSettings) -> dict[str, Any]:
async def _update_settings(self, delta: TTSSettings) -> TTSSettings:
"""Apply a settings delta with voice validation.
Args:
delta: Settings delta. Can include 'voice', 'prompt', etc.
Returns:
Dict mapping changed field names to their previous values.
A delta-mode settings object whose given fields are the ones that
actually changed (values are the *previous* values).
"""
if is_given(delta.voice) and delta.voice not in self.AVAILABLE_VOICES:
logger.warning(f"Voice '{delta.voice}' not in known voices list. Using anyway.")

6
src/pipecat/services/gradium/stt.py
View File

@@ -203,17 +203,17 @@ class GradiumSTTService(WebsocketSTTService):
"""
return True
async def _update_settings(self, delta: STTSettings) -> dict[str, Any]:
async def _update_settings(self, delta: STTSettings) -> Settings:
"""Apply a settings delta, sync params, and reconnect.
Args:
delta: A :class:`STTSettings` (or ``GradiumSTTService.Settings``) delta.
Returns:
Dict mapping changed field names to their previous values.
A delta-mode settings object with only the changed fields set.
"""
changed = await super()._update_settings(delta)
if not changed:
if not changed.given_fields():
return changed
if self._websocket:

10
src/pipecat/services/gradium/tts.py
View File

@@ -21,7 +21,7 @@ from pipecat.frames.frames import (
TTSAudioRawFrame,
TTSStoppedFrame,
)
from pipecat.services.settings import TTSSettings
from pipecat.services.settings import TTSSettings, is_given
from pipecat.services.tts_service import WebsocketTTSService
from pipecat.utils.tracing.service_decorators import traced_tts
@@ -149,21 +149,21 @@ class GradiumTTSService(WebsocketTTSService):
"""
return True
async def _update_settings(self, delta: TTSSettings) -> dict[str, Any]:
async def _update_settings(self, delta: TTSSettings) -> Settings:
"""Apply a settings delta and reconnect if voice changed.
Args:
delta: A :class:`TTSSettings` (or ``GradiumTTSService.Settings``) delta.
Returns:
Dict mapping changed field names to their previous values.
A delta-mode settings object with only the changed fields set.
"""
changed = await super()._update_settings(delta)
if "voice" in changed:
if is_given(changed.voice):
await self._disconnect()
await self._connect()
else:
self._warn_unhandled_updated_settings(changed)
self._warn_unhandled_updated_settings(changed.given_fields())
return changed
def _build_msg(self, text: str = "", context_id: str = "") -> dict:

18
src/pipecat/services/grok/realtime/llm.py
View File

@@ -118,7 +118,9 @@ class GrokRealtimeLLMSettings(LLMSettings):
# -- apply_update override -----------------------------------------------
def apply_update(self, delta: "GrokRealtimeLLMService.Settings") -> Dict[str, Any]:
def apply_update(
self, delta: "GrokRealtimeLLMService.Settings"
) -> "GrokRealtimeLLMService.Settings":
"""Merge a delta, keeping ``system_instruction`` in sync with SP.
When the delta contains ``session_properties``, it **replaces** the
@@ -131,13 +133,13 @@ class GrokRealtimeLLMSettings(LLMSettings):
# 2. SP → top-level: if the SP was just replaced and carries
# instructions that the delta didn't set at top level, pull it up.
if "session_properties" in changed and is_given(self.session_properties):
if is_given(changed.session_properties) and is_given(self.session_properties):
sp = self.session_properties
if "system_instruction" not in changed and sp.instructions is not None:
if not is_given(changed.system_instruction) 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
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,
@@ -582,7 +584,7 @@ class GrokRealtimeLLMService(LLMService):
return
await self.push_error(error_msg=f"Error sending client event: {e}", exception=e)
async def _update_settings(self, delta):
async def _update_settings(self, delta: Settings) -> Settings:
"""Apply a settings delta, sending a session update when needed."""
# Capture audio config before the update — a wholesale SP replacement
# would lose it since the new SP likely has audio=None.
@@ -592,13 +594,13 @@ class GrokRealtimeLLMService(LLMService):
changed = await super()._update_settings(delta)
# Re-establish audio config if it was lost during SP replacement.
if "session_properties" in changed and input_rate and output_rate:
if is_given(changed.session_properties) and input_rate and output_rate:
self._ensure_audio_config(input_rate, output_rate)
handled = {"session_properties", "system_instruction"}
if changed.keys() & handled:
if changed.given_fields().keys() & handled:
await self._send_session_update()
self._warn_unhandled_updated_settings(changed.keys() - handled)
self._warn_unhandled_updated_settings(changed.given_fields().keys() - handled)
return changed
async def _send_session_update(self):

4
src/pipecat/services/inworld/tts.py
View File

@@ -826,14 +826,14 @@ class InworldTTSService(WebsocketTTSService):
await self._disconnect_websocket()
async def _update_settings(self, delta: TTSSettings) -> dict[str, Any]:
async def _update_settings(self, delta: TTSSettings) -> TTSSettings:
"""Apply a settings delta.
Settings are stored but not applied to the active connection.
"""
changed = await super()._update_settings(delta)
if not changed:
if not changed.given_fields():
return changed
await self._disconnect()

15
src/pipecat/services/llm_service.py
View File

@@ -59,7 +59,7 @@ from pipecat.processors.aggregators.llm_response import (
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.ai_service import AIService
from pipecat.services.settings import LLMSettings
from pipecat.services.settings import LLMSettings, is_given
from pipecat.turns.user_turn_completion_mixin import UserTurnCompletionLLMServiceMixin
from pipecat.utils.context.llm_context_summarization import (
DEFAULT_SUMMARIZATION_TIMEOUT,
@@ -345,18 +345,19 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
else:
self._settings.system_instruction = completion_instructions
async def _update_settings(self, delta: LLMSettings) -> dict[str, Any]:
async def _update_settings(self, delta: LLMSettings) -> LLMSettings:
"""Apply a settings delta, handling turn-completion fields.
Args:
delta: An LLM settings delta.
Returns:
Dict mapping changed field names to their previous values.
A delta-mode ``LLMSettings`` with changed fields set to their
previous values.
"""
changed = await super()._update_settings(delta)
if "filter_incomplete_user_turns" in changed:
if is_given(changed.filter_incomplete_user_turns):
self._filter_incomplete_user_turns = (
self._settings.filter_incomplete_user_turns or False
)
@@ -373,14 +374,14 @@ class LLMService(UserTurnCompletionLLMServiceMixin, AIService):
self._settings.system_instruction = self._base_system_instruction
self._base_system_instruction = None
if "user_turn_completion_config" in changed and self._filter_incomplete_user_turns:
if is_given(changed.user_turn_completion_config) and self._filter_incomplete_user_turns:
self.set_user_turn_completion_config(self._settings.user_turn_completion_config)
self._compose_system_instruction()
if (
"system_instruction" in changed
is_given(changed.system_instruction)
and self._filter_incomplete_user_turns
and "filter_incomplete_user_turns" not in changed
and not is_given(changed.filter_incomplete_user_turns)
):
# system_instruction changed while turn completion is active.
# Treat the new value as the new base and recompose.

6
src/pipecat/services/lmnt/tts.py
View File

@@ -233,18 +233,18 @@ class LmntTTSService(InterruptibleTTSService):
await self._disconnect_websocket()
async def _update_settings(self, delta: TTSSettings) -> dict[str, Any]:
async def _update_settings(self, delta: TTSSettings) -> Settings:
"""Apply a settings delta.
Args:
delta: A :class:`TTSSettings` (or ``LmntTTSService.Settings``) delta.
Returns:
Dict mapping changed field names to their previous values.
A delta-mode settings object with only the changed fields set.
"""
changed = await super()._update_settings(delta)
if changed:
if changed.given_fields():
await self._disconnect()
await self._connect()

4
src/pipecat/services/neuphonic/tts.py
View File

@@ -212,10 +212,10 @@ class NeuphonicTTSService(InterruptibleTTSService):
"""
return language_to_neuphonic_lang_code(language)
async def _update_settings(self, delta: TTSSettings) -> dict[str, Any]:
async def _update_settings(self, delta: TTSSettings) -> Settings:
"""Apply a settings delta and reconnect with new configuration."""
changed = await super()._update_settings(delta)
if changed:
if changed.given_fields():
await self._disconnect()
await self._connect()
logger.info(f"Switching TTS to settings: [{self._settings}]")

6
src/pipecat/services/nvidia/stt.py
View File

@@ -637,18 +637,18 @@ class NvidiaSegmentedSTTService(SegmentedSTTService):
self._config = self._create_recognition_config()
logger.debug(f"Initialized NvidiaSegmentedSTTService with model: {self._settings.model}")
async def _update_settings(self, delta: STTSettings) -> dict[str, Any]:
async def _update_settings(self, delta: STTSettings) -> Settings:
"""Apply a settings delta and sync internal state.
Args:
delta: A :class:`STTSettings` (or ``NvidiaSegmentedSTTService.Settings``) delta.
Returns:
Dict mapping changed field names to their previous values.
A delta-mode settings object with only the changed fields set.
"""
changed = await super()._update_settings(delta)
if changed:
if changed.given_fields():
self._config = self._create_recognition_config()
return changed

6
src/pipecat/services/nvidia/tts.py
View File

@@ -186,16 +186,16 @@ class NvidiaTTSService(TTSService):
stacklevel=2,
)
async def _update_settings(self, delta: Settings) -> dict[str, Any]:
async def _update_settings(self, delta: Settings) -> Settings:
"""Apply a settings delta.
Settings are stored but not applied to the active connection.
"""
changed = await super()._update_settings(delta)
if not changed:
if not changed.given_fields():
return changed
# TODO: reconnect gRPC client to apply changed settings.
self._warn_unhandled_updated_settings(changed)
self._warn_unhandled_updated_settings(changed.given_fields())
return changed
def _initialize_client(self):

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

@@ -126,7 +126,9 @@ class OpenAIRealtimeLLMSettings(LLMSettings):
# -- apply_update override -----------------------------------------------
def apply_update(self, delta: "OpenAIRealtimeLLMService.Settings") -> Dict[str, Any]:
def apply_update(
self, delta: "OpenAIRealtimeLLMService.Settings"
) -> "OpenAIRealtimeLLMService.Settings":
"""Merge a delta, keeping ``model``/``system_instruction`` in sync with SP.
When the delta contains ``session_properties``, it **replaces** the
@@ -140,18 +142,18 @@ class OpenAIRealtimeLLMSettings(LLMSettings):
# 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):
if is_given(changed.session_properties) and is_given(self.session_properties):
sp = self.session_properties
if "model" not in changed and sp.model is not None:
if not is_given(changed.model) 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:
changed.model = old_model
if not is_given(changed.system_instruction) 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
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,
@@ -673,13 +675,13 @@ class OpenAIRealtimeLLMService(LLMService):
# treat a send-side error as fatal.
await self.push_error(error_msg=f"Error sending client event: {e}", exception=e)
async def _update_settings(self, delta):
async def _update_settings(self, delta: Settings) -> Settings:
"""Apply a settings delta, sending a session update when needed."""
changed = await super()._update_settings(delta)
handled = {"session_properties", "system_instruction"}
if changed.keys() & handled:
if changed.given_fields().keys() & handled:
await self._send_session_update()
self._warn_unhandled_updated_settings(changed.keys() - handled)
self._warn_unhandled_updated_settings(changed.given_fields().keys() - handled)
return changed
async def _send_session_update(self):

9
src/pipecat/services/openai/stt.py
View File

@@ -17,7 +17,7 @@ Provides two STT services:
import base64
import json
from dataclasses import dataclass, field
from typing import Any, AsyncGenerator, Literal, Optional, Union
from typing import AsyncGenerator, Literal, Optional, Union
from loguru import logger
@@ -369,7 +369,7 @@ class OpenAIRealtimeSTTService(WebsocketSTTService):
"""
return True
async def _update_settings(self, delta: STTSettings) -> dict[str, Any]:
async def _update_settings(self, delta: STTSettings) -> Settings:
"""Apply a settings delta and send session update if needed.
Sends a ``session.update`` to the server when the session is active.
@@ -378,11 +378,12 @@ class OpenAIRealtimeSTTService(WebsocketSTTService):
delta: A :class:`STTSettings` (or ``OpenAIRealtimeSTTService.Settings``) delta.
Returns:
Dict mapping changed field names to their previous values.
A delta-mode settings object whose ``given_fields()`` contains
only the fields that actually changed.
"""
changed = await super()._update_settings(delta)
if changed and self._session_ready:
if changed.given_fields() and self._session_ready:
await self._send_session_update()
return changed

4
src/pipecat/services/openai_realtime_beta/openai.py
View File

@@ -503,10 +503,10 @@ class OpenAIRealtimeBetaLLMService(LLMService):
# treat a send-side error as fatal.
await self.push_error(error_msg=f"Error sending client event: {e}", exception=e)
async def _update_settings(self, delta):
async def _update_settings(self, delta: Settings) -> Settings:
"""Apply a settings delta."""
changed = await super()._update_settings(delta)
self._warn_unhandled_updated_settings(changed.keys())
self._warn_unhandled_updated_settings(changed.given_fields().keys())
return changed
async def _send_session_update(self):

6
src/pipecat/services/piper/tts.py
View File

@@ -121,16 +121,16 @@ class PiperTTSService(TTSService):
"""
return True
async def _update_settings(self, delta: Settings) -> dict[str, Any]:
async def _update_settings(self, delta: Settings) -> Settings:
"""Apply a settings delta.
Settings are stored but not applied to the active connection.
"""
changed = await super()._update_settings(delta)
if not changed:
if not changed.given_fields():
return changed
# TODO: voice changes would require re-downloading and loading the model.
self._warn_unhandled_updated_settings(changed)
self._warn_unhandled_updated_settings(changed.given_fields())
return changed
@traced_tts

8
src/pipecat/services/rime/tts.py
View File

@@ -394,7 +394,7 @@ class RimeTTSService(WebsocketTTSService):
self._extra_msg_fields["inlineSpeedAlpha"] = ",".join(speed_vals + [str(speed)])
return f"[{text}]"
async def _update_settings(self, delta: TTSSettings) -> dict[str, Any]:
async def _update_settings(self, delta: TTSSettings) -> TTSSettings:
"""Apply a settings delta and reconnect if necessary.
Since all settings are WebSocket URL query parameters,
@@ -402,7 +402,7 @@ class RimeTTSService(WebsocketTTSService):
"""
changed = await super()._update_settings(delta)
if changed and self._websocket:
if changed.given_fields() and self._websocket:
await self._disconnect()
await self._connect()
@@ -1199,7 +1199,7 @@ class RimeNonJsonTTSService(InterruptibleTTSService):
except Exception as e:
yield ErrorFrame(error=f"Unknown error occurred: {e}")
async def _update_settings(self, delta: TTSSettings) -> dict[str, Any]:
async def _update_settings(self, delta: TTSSettings) -> TTSSettings:
"""Apply a settings delta and reconnect if necessary.
Since all settings are WebSocket URL query parameters,
@@ -1207,7 +1207,7 @@ class RimeNonJsonTTSService(InterruptibleTTSService):
"""
changed = await super()._update_settings(delta)
if changed:
if changed.given_fields():
logger.debug("Settings changed, reconnecting WebSocket with new parameters")
await self._disconnect()
await self._connect()

10
src/pipecat/services/sarvam/stt.py
View File

@@ -13,7 +13,7 @@ can handle multiple audio formats for Indian language speech recognition.
import base64
from dataclasses import dataclass, field
from typing import Any, AsyncGenerator, Dict, Literal, Optional
from typing import AsyncGenerator, Dict, Literal, Optional
from loguru import logger
from pydantic import BaseModel
@@ -369,14 +369,14 @@ class SarvamSTTService(STTService):
if self._socket_client:
await self._socket_client.flush()
async def _update_settings(self, delta: STTSettings) -> dict[str, Any]:
async def _update_settings(self, delta: STTSettings) -> STTSettings:
"""Apply a settings delta, validate, sync state, and reconnect.
Args:
delta: A :class:`STTSettings` (or ``SarvamSTTService.Settings``) delta.
Returns:
Dict mapping changed field names to their previous values.
A delta-mode settings object with changed fields.
Raises:
ValueError: If a setting is not supported by the current model.
@@ -398,11 +398,11 @@ class SarvamSTTService(STTService):
# Language and prompt are WebSocket connect-time parameters; reconnect to apply.
reconnect_fields = {"language", "prompt"}
if changed.keys() & reconnect_fields:
if changed.given_fields().keys() & reconnect_fields:
await self._disconnect()
await self._connect()
unhandled = {k: v for k, v in changed.items() if k not in reconnect_fields}
unhandled = {k: v for k, v in changed.given_fields().items() if k not in reconnect_fields}
if unhandled:
self._warn_unhandled_updated_settings(unhandled)

6
src/pipecat/services/sarvam/tts.py
View File

@@ -42,7 +42,7 @@ import base64
import json
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, AsyncGenerator, ClassVar, Dict, List, Optional, Tuple
from typing import AsyncGenerator, ClassVar, Dict, List, Optional, Tuple
import aiohttp
from loguru import logger
@@ -1060,11 +1060,11 @@ class SarvamTTSService(InterruptibleTTSService):
if isinstance(frame, (LLMFullResponseEndFrame, EndFrame)):
await self.flush_audio()
async def _update_settings(self, delta: TTSSettings) -> dict[str, Any]:
async def _update_settings(self, delta: TTSSettings) -> TTSSettings:
"""Apply a settings delta and resend config if voice changed."""
changed = await super()._update_settings(delta)
if changed:
if changed.given_fields():
await self._send_config()
return changed

23
src/pipecat/services/settings.py
View File

@@ -190,7 +190,7 @@ class ServiceSettings:
result.update(self.extra)
return result
def apply_update(self: _S, delta: _S) -> Dict[str, Any]:
def apply_update(self: _S, delta: _S) -> _S:
"""Merge a delta-mode object into this store-mode object.
Only fields in *delta* that are **given** (i.e. not ``NOT_GIVEN``)
@@ -204,9 +204,11 @@ class ServiceSettings:
delta: A delta-mode settings object of the same type.
Returns:
A dict mapping each changed field name to its **pre-update** value.
Use ``changed.keys()`` for the set of names, or index with
``changed["field"]`` to inspect the old value.
A delta-mode settings instance of the same type, with changed
fields set to their **pre-update** (old) values and unchanged
fields left as ``NOT_GIVEN``. Use ``changed.given_fields()``
for the set of changed names, or access fields directly with
typed attribute access (e.g. ``is_given(changed.voice)``).
Examples::
@@ -215,10 +217,10 @@ class ServiceSettings:
# delta-mode object (only voice is set)
delta = TTSSettings(voice="bob")
changed = current.apply_update(delta)
# changed == {"voice": "alice"}
# is_given(changed.voice) is True, changed.voice == "alice"
# current.voice == "bob", current.language == "en"
"""
changed: Dict[str, Any] = {}
changed: _S = type(self)() # delta-mode: all NOT_GIVEN
for f in fields(self):
if f.name == "extra":
continue
@@ -228,14 +230,19 @@ class ServiceSettings:
old_val = getattr(self, f.name)
if old_val != new_val:
setattr(self, f.name, new_val)
changed[f.name] = old_val
# Store the old value. Use None as a stand-in when the old
# value was NOT_GIVEN, since NOT_GIVEN is the "unset"
# sentinel in delta mode and would hide the change. In
# store-mode objects (the normal case) this never triggers
# because validate_complete() ensures no NOT_GIVEN fields.
setattr(changed, f.name, None if isinstance(old_val, _NotGiven) else old_val)
# Merge extra
for key, new_val in delta.extra.items():
old_val = self.extra.get(key, NOT_GIVEN)
if old_val != new_val:
self.extra[key] = new_val
changed[key] = old_val
changed.extra[key] = old_val
return changed

8
src/pipecat/services/soniox/stt.py
View File

@@ -9,7 +9,7 @@
import json
import time
from dataclasses import dataclass, field
from typing import Any, AsyncGenerator, List, Optional
from typing import AsyncGenerator, List, Optional
from loguru import logger
from pydantic import BaseModel
@@ -297,18 +297,18 @@ class SonioxSTTService(WebsocketSTTService):
await super().start(frame)
await self._connect()
async def _update_settings(self, delta: Settings) -> dict[str, Any]:
async def _update_settings(self, delta: Settings) -> STTSettings:
"""Apply settings delta and reconnect if anything changed.
Args:
delta: A settings delta.
Returns:
Dict mapping changed field names to their previous values.
A delta-mode settings object with changed fields.
"""
changed = await super()._update_settings(delta)
if changed:
if changed.given_fields():
await self._disconnect()
await self._connect()

23
src/pipecat/services/speechmatics/stt.py
View File

@@ -537,7 +537,7 @@ class SpeechmaticsSTTService(STTService):
await super().start(frame)
await self._connect()
async def _update_settings(self, delta: Settings) -> dict[str, Any]:
async def _update_settings(self, delta: Settings) -> Settings:
"""Apply settings delta, reconnecting only when necessary.
Fields are classified into three categories (see
@@ -554,25 +554,28 @@ class SpeechmaticsSTTService(STTService):
delta: A settings delta.
Returns:
Dict mapping changed field names to their previous values.
A delta-mode settings object whose given fields are the ones that
actually changed (values are the *previous* values).
"""
changed = await super()._update_settings(delta)
if not changed:
if not changed.given_fields():
return changed
no_reconnect = self.Settings.HOT_FIELDS | self.Settings.LOCAL_FIELDS
needs_reconnect = bool(changed.keys() - no_reconnect)
needs_reconnect = bool(changed.given_fields().keys() - no_reconnect)
if needs_reconnect:
logger.debug(f"{self} settings update requires reconnect: {changed.keys()}")
logger.debug(
f"{self} settings update requires reconnect: {changed.given_fields().keys()}"
)
# Connection-level fields changed — rebuild the SDK config
# from the now-updated self._settings, then reconnect.
self._config = self._build_config(self._settings)
await self._disconnect()
await self._connect()
elif changed.keys() & self.Settings.HOT_FIELDS:
logger.debug(f"{self} applying hot settings update: {changed.keys()}")
elif changed.given_fields().keys() & self.Settings.HOT_FIELDS:
logger.debug(f"{self} applying hot settings update: {changed.given_fields().keys()}")
if self._config.enable_diarization:
# Only hot-updatable fields changed — push to the live session.
self._config.speaker_config.focus_speakers = self._settings.focus_speakers
@@ -582,13 +585,13 @@ class SpeechmaticsSTTService(STTService):
self._client.update_diarization_config(self._config.speaker_config)
else:
logger.debug(
f"{self} hot settings updated but diarization not enabled: {changed.keys()}. ignoring."
f"{self} hot settings updated but diarization not enabled: {changed.given_fields().keys()}. ignoring."
)
# Diarization not enabled — the new settings will take effect
# if/when diarization is enabled, which does require a reconnect.
elif changed.keys() & self.Settings.LOCAL_FIELDS:
elif changed.given_fields().keys() & self.Settings.LOCAL_FIELDS:
logger.debug(
f"{self} local settings update, no special action required: {changed.keys()}"
f"{self} local settings update, no special action required: {changed.given_fields().keys()}"
)
# Only local fields changed — no need to push to the STT engine,
# the new settings will take effect immediately.

7
src/pipecat/services/stt_service.py
View File

@@ -270,20 +270,21 @@ class STTService(AIService):
await self._cancel_ttfb_timeout()
await self._cancel_keepalive_task()
async def _update_settings(self, delta: STTSettings) -> dict[str, Any]:
async def _update_settings(self, delta: STTSettings) -> STTSettings:
"""Apply an STT settings delta.
Handles ``model`` (via parent). Translates ``Language`` enum values
before applying so the stored value is a service-specific string.
Concrete services should override this method and handle language
changes (including any reconnect logic) based on the returned
changed-field dict.
changed-field object.
Args:
delta: An STT settings delta.
Returns:
Dict mapping changed field names to their previous values.
A delta-mode ``STTSettings`` with changed fields set to their
previous values.
"""
# Translate language *before* applying so the stored value is canonical
if is_given(delta.language) and isinstance(delta.language, Language):

5
src/pipecat/services/tts_service.py
View File

@@ -591,7 +591,7 @@ class TTSService(AIService):
if not (agg_type == aggregation_type and func == transform_function)
]
async def _update_settings(self, delta: TTSSettings) -> dict[str, Any]:
async def _update_settings(self, delta: TTSSettings) -> TTSSettings:
"""Apply a TTS settings delta.
Translates language to service-specific value before applying.
@@ -600,7 +600,8 @@ class TTSService(AIService):
delta: A TTS settings delta.
Returns:
Dict mapping changed field names to their previous values.
A delta-mode ``TTSSettings`` with changed fields set to their
previous values.
"""
# Translate language *before* applying so the stored value is canonical
if is_given(delta.language) and isinstance(delta.language, Language):

8
src/pipecat/services/ultravox/llm.py
View File

@@ -57,7 +57,7 @@ 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 NOT_GIVEN, LLMSettings, _NotGiven
from pipecat.services.settings import NOT_GIVEN, LLMSettings, _NotGiven, is_given
from pipecat.utils.time import time_now_iso8601
try:
@@ -383,11 +383,11 @@ class UltravoxRealtimeLLMService(LLMService):
await self.cancel_task(self._receive_task, timeout=1.0)
self._receive_task = None
async def _update_settings(self, delta: Settings):
async def _update_settings(self, delta: Settings) -> Settings:
changed = await super()._update_settings(delta)
if "output_medium" in changed:
if is_given(changed.output_medium):
await self._update_output_medium(self._settings.output_medium)
self._warn_unhandled_updated_settings(changed.keys() - {"output_medium"})
self._warn_unhandled_updated_settings(changed.given_fields().keys() - {"output_medium"})
return changed
#

80
tests/test_settings.py
View File

@@ -102,8 +102,8 @@ class TestApplyUpdate:
current = TTSSettings(voice="alice", language="en")
delta = TTSSettings(voice="bob")
changed = current.apply_update(delta)
assert changed.keys() == {"voice"}
assert changed["voice"] == "alice" # old value
assert set(changed.given_fields()) == {"voice"}
assert changed.voice == "alice" # old value
assert current.voice == "bob"
assert current.language == "en"
@@ -111,14 +111,14 @@ class TestApplyUpdate:
current = TTSSettings(voice="alice", language="en")
delta = TTSSettings(voice="alice")
changed = current.apply_update(delta)
assert changed == {}
assert changed.given_fields() == {}
assert current.voice == "alice"
def test_apply_update_not_given_skipped(self):
current = TTSSettings(voice="alice", language="en")
delta = TTSSettings() # all NOT_GIVEN
changed = current.apply_update(delta)
assert changed == {}
assert changed.given_fields() == {}
assert current.voice == "alice"
assert current.language == "en"
@@ -126,9 +126,9 @@ class TestApplyUpdate:
current = LLMSettings(temperature=0.7, max_tokens=100)
delta = LLMSettings(temperature=0.9, max_tokens=200, top_p=0.95)
changed = current.apply_update(delta)
assert changed.keys() == {"temperature", "max_tokens", "top_p"}
assert changed["temperature"] == 0.7
assert changed["max_tokens"] == 100
assert set(changed.given_fields()) == {"temperature", "max_tokens", "top_p"}
assert changed.temperature == 0.7
assert changed.max_tokens == 100
assert current.temperature == 0.9
assert current.max_tokens == 200
assert current.top_p == 0.95
@@ -139,8 +139,8 @@ class TestApplyUpdate:
delta = TTSSettings()
delta.extra = {"speed": 1.2}
changed = current.apply_update(delta)
assert "speed" in changed
assert changed["speed"] == 1.0 # old value
assert "speed" in changed.extra
assert changed.extra["speed"] == 1.0 # old value
assert current.extra == {"speed": 1.2, "stability": 0.5}
def test_apply_update_extra_no_change(self):
@@ -149,14 +149,14 @@ class TestApplyUpdate:
delta = TTSSettings()
delta.extra = {"speed": 1.0}
changed = current.apply_update(delta)
assert changed == {}
assert changed.given_fields() == {}
def test_apply_update_model_field(self):
current = ServiceSettings(model="old-model")
delta = ServiceSettings(model="new-model")
changed = current.apply_update(delta)
assert changed.keys() == {"model"}
assert changed["model"] == "old-model"
assert set(changed.given_fields()) == {"model"}
assert changed.model == "old-model"
assert current.model == "new-model"
def test_apply_update_none_is_a_valid_value(self):
@@ -164,15 +164,15 @@ class TestApplyUpdate:
current = TTSSettings()
delta = TTSSettings(language=None)
changed = current.apply_update(delta)
assert "language" in changed
assert is_given(changed.language)
assert current.language is None
def test_apply_update_none_to_value(self):
current = TTSSettings(language=None)
delta = TTSSettings(language="en")
changed = current.apply_update(delta)
assert "language" in changed
assert changed["language"] is None # old value was None
assert is_given(changed.language)
assert changed.language is None # old value was None
assert current.language == "en"
@@ -301,9 +301,9 @@ class TestRoundtrip:
delta = TTSSettings.from_mapping(raw)
changed = current.apply_update(delta)
assert changed.keys() == {"voice", "speed"}
assert changed["voice"] == "alice"
assert changed["speed"] == 1.0
assert set(changed.given_fields()) == {"voice", "speed"}
assert changed.voice == "alice"
assert changed.extra["speed"] == 1.0
assert current.voice == "bob"
assert current.language == "en"
assert current.extra["speed"] == 1.2
@@ -313,8 +313,8 @@ class TestRoundtrip:
current = LLMSettings(model="gpt-4o", temperature=0.7)
delta = LLMSettings.from_mapping({"model": "gpt-4o-mini", "temperature": 0.9})
changed = current.apply_update(delta)
assert changed.keys() == {"model", "temperature"}
assert changed["model"] == "gpt-4o"
assert set(changed.given_fields()) == {"model", "temperature"}
assert changed.model == "gpt-4o"
assert current.model == "gpt-4o-mini"
assert current.temperature == 0.9
@@ -348,7 +348,7 @@ class TestDeepgramSTTSettingsApplyUpdate:
changed = current.apply_update(delta)
assert current.punctuate is False
assert "punctuate" in changed
assert is_given(changed.punctuate)
# Other fields are untouched
assert current.model == "nova-3-general"
assert current.language == "en"
@@ -362,7 +362,7 @@ class TestDeepgramSTTSettingsApplyUpdate:
changed = current.apply_update(delta)
assert current.model == "nova-2"
assert "model" in changed
assert is_given(changed.model)
def test_apply_update_language(self):
"""language field is updated directly."""
@@ -373,14 +373,14 @@ class TestDeepgramSTTSettingsApplyUpdate:
changed = current.apply_update(delta)
assert current.language == "es"
assert "language" in changed
assert is_given(changed.language)
def test_apply_update_no_change(self):
"""Delta with same values should report no changes."""
current = self._make_store()
delta = DeepgramSTTSettings(punctuate=True)
changed = current.apply_update(delta)
assert changed == {}
assert changed.given_fields() == {}
def test_apply_update_multiple_fields(self):
"""Multiple flat fields updated at once."""
@@ -392,7 +392,7 @@ class TestDeepgramSTTSettingsApplyUpdate:
assert current.model == "nova-2"
assert current.language == "fr"
assert current.punctuate is False
assert changed.keys() == {"model", "language", "punctuate"}
assert set(changed.given_fields()) == {"model", "language", "punctuate"}
class TestDeepgramSTTSettingsFromMapping:
@@ -441,7 +441,7 @@ class TestDeepgramSTTSettingsFromMapping:
assert current.diarize is True
# Unchanged fields stay put
assert current.model == "nova-3-general"
assert "punctuate" in changed
assert is_given(changed.punctuate)
def test_roundtrip_model_via_dict(self):
"""Dict update with model should change top-level model field."""
@@ -455,7 +455,7 @@ class TestDeepgramSTTSettingsFromMapping:
changed = current.apply_update(delta)
assert current.model == "nova-2"
assert "model" in changed
assert is_given(changed.model)
# ---------------------------------------------------------------------------
@@ -475,7 +475,7 @@ class TestDeepgramSageMakerSTTSettings:
assert store.model == "nova-2"
assert store.language == "en"
assert "model" in changed
assert is_given(changed.model)
# ---------------------------------------------------------------------------
@@ -650,7 +650,7 @@ class TestOpenAIRealtimeSettingsApplyUpdate:
delta = OpenAIRealtimeLLMSettings(model="gpt-realtime-2.0")
changed = store.apply_update(delta)
assert "model" in changed
assert is_given(changed.model)
assert store.model == "gpt-realtime-2.0"
assert store.session_properties.model == "gpt-realtime-2.0"
@@ -660,7 +660,7 @@ class TestOpenAIRealtimeSettingsApplyUpdate:
delta = OpenAIRealtimeLLMSettings(system_instruction="Be helpful.")
changed = store.apply_update(delta)
assert "system_instruction" in changed
assert is_given(changed.system_instruction)
assert store.system_instruction == "Be helpful."
assert store.session_properties.instructions == "Be helpful."
@@ -678,7 +678,7 @@ class TestOpenAIRealtimeSettingsApplyUpdate:
delta = OpenAIRealtimeLLMSettings(session_properties=new_sp)
changed = store.apply_update(delta)
assert "session_properties" in changed
assert is_given(changed.session_properties)
assert store.session_properties.output_modalities == ["text"]
# Fields not in the new SP become None (wholesale replacement)
# But model is synced from top-level
@@ -691,7 +691,7 @@ class TestOpenAIRealtimeSettingsApplyUpdate:
delta = OpenAIRealtimeLLMSettings(session_properties=new_sp)
changed = store.apply_update(delta)
assert "model" in changed
assert is_given(changed.model)
assert store.model == "gpt-realtime-2.0"
assert store.session_properties.model == "gpt-realtime-2.0"
@@ -702,7 +702,7 @@ class TestOpenAIRealtimeSettingsApplyUpdate:
delta = OpenAIRealtimeLLMSettings(session_properties=new_sp)
changed = store.apply_update(delta)
assert "system_instruction" in changed
assert is_given(changed.system_instruction)
assert store.system_instruction == "New instructions."
assert store.session_properties.instructions == "New instructions."
@@ -740,7 +740,7 @@ class TestOpenAIRealtimeSettingsApplyUpdate:
delta = OpenAIRealtimeLLMSettings(temperature=0.5)
changed = store.apply_update(delta)
assert "temperature" in changed
assert is_given(changed.temperature)
assert store.temperature == 0.5
# SP should be untouched (same object)
assert store.session_properties is original_sp
@@ -811,7 +811,7 @@ class TestOpenAIRealtimeSettingsFromMapping:
delta = OpenAIRealtimeLLMSettings.from_mapping(raw)
changed = store.apply_update(delta)
assert "session_properties" in changed
assert is_given(changed.session_properties)
assert store.session_properties.instructions == "Be concise."
assert store.session_properties.output_modalities == ["text"]
assert store.system_instruction == "Be concise."
@@ -848,7 +848,7 @@ class TestGrokRealtimeSettingsApplyUpdate:
delta = GrokRealtimeLLMSettings(system_instruction="Be helpful.")
changed = store.apply_update(delta)
assert "system_instruction" in changed
assert is_given(changed.system_instruction)
assert store.system_instruction == "Be helpful."
assert store.session_properties.instructions == "Be helpful."
@@ -866,7 +866,7 @@ class TestGrokRealtimeSettingsApplyUpdate:
delta = GrokRealtimeLLMSettings(session_properties=new_sp)
changed = store.apply_update(delta)
assert "session_properties" in changed
assert is_given(changed.session_properties)
assert store.session_properties.voice == "Sal"
# instructions is synced from top-level system_instruction
assert store.session_properties.instructions == "Old instructions."
@@ -878,7 +878,7 @@ class TestGrokRealtimeSettingsApplyUpdate:
delta = GrokRealtimeLLMSettings(session_properties=new_sp)
changed = store.apply_update(delta)
assert "system_instruction" in changed
assert is_given(changed.system_instruction)
assert store.system_instruction == "New instructions."
assert store.session_properties.instructions == "New instructions."
@@ -906,7 +906,7 @@ class TestGrokRealtimeSettingsApplyUpdate:
delta = GrokRealtimeLLMSettings(temperature=0.5)
changed = store.apply_update(delta)
assert "temperature" in changed
assert is_given(changed.temperature)
assert store.temperature == 0.5
# SP should be untouched (same object)
assert store.session_properties is original_sp
@@ -977,7 +977,7 @@ class TestGrokRealtimeSettingsFromMapping:
delta = GrokRealtimeLLMSettings.from_mapping(raw)
changed = store.apply_update(delta)
assert "session_properties" in changed
assert is_given(changed.session_properties)
assert store.session_properties.instructions == "Be concise."
assert store.session_properties.voice == "Eve"
assert store.system_instruction == "Be concise."