Broad service settings refactor, with the primary aim of making service settings discoverable and strongly-typed. Service settings can be updated at runtime with *UpdateSettingsFrames.

Does not (yet) touch `InputParams`, to avoid scope creep and touching something currently part of the public API. But there is a lot of overlap between `*Settings` object fields and `InputParams` fields.

Other than discoverability/typing, these are some other improvements brought by this refactor:
- There is now a single code path (see `_update_settings_from_typed`) where services can respond to settings changes (by, say, reconnecting if needed), improving maintainability and guaranteeing one and only one reconnection no matter which settings changed
- `set_language`/`set_model`/`set_voice`—which we're assuming are usable as public methods, though *not* recommended over `*UpdateSettingsFrame`—all use the same code path as settings updates. They're also now all consistent in that, if a service needs to respond to a change (by, say, reconnecting if needed), any of these methods will kick off that process. Note that this is technically a behavior change.
- Several services now properly react to changed settings by reconnecting:
  - `AWSTranscribeSTTService`
  - `AzureSTTService`
  - `SonioxSTTService`
  - `GladiaSTTService`
  - `SpeechmaticsSTTService`
  - `AssemblyAISTTService`
  - `CartesiaSTTService`
  - `FishAudioTTSService` (would previously only reconnect when `model` changed)
  - `GoogleSTTService`
  - `SpeechmaticsSTTService` (which previously only handled *some* settings updates through a nonstandard public `update_params` method)
  - `GradiumSTTService`
  - `NvidiaSegmentedSTTService` (which previously only handled changes to language)
- Bookkeeping across various services has been reduced, mostly by deduping ivars; the `self._settings` ivar is treated as the source of truth

NOTE: I pretty much guarantee that there are services missed in this PR in terms of bringing to consistency with how updates are handled (like whether changes in certain fields trigger reconnects when they need to). We can squash remaining inconsistencies as we stumble onto them, service by service. The goal here is to get things *mostly* in order, and establish the infrastructure and patterns we'll need going forward.

src/pipecat/services/gradium/stt.py

@@ -12,6 +12,7 @@ WebSocket API for streaming audio transcription.
 
 import base64
 import json
+from dataclasses import dataclass, field
 from typing import AsyncGenerator, Optional
 
 from loguru import logger
@@ -27,6 +28,7 @@ from pipecat.frames.frames import (
     VADUserStoppedSpeakingFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
+from pipecat.services.settings import NOT_GIVEN, STTSettings, is_given
 from pipecat.services.stt_latency import GRADIUM_TTFS_P99
 from pipecat.services.stt_service import WebsocketSTTService
 from pipecat.transcriptions.language import Language, resolve_language
@@ -64,6 +66,18 @@ def language_to_gradium_language(language: Language) -> Optional[str]:
     return resolve_language(language, LANGUAGE_MAP, use_base_code=True)
 
 
+@dataclass
+class GradiumSTTSettings(STTSettings):
+    """Typed settings for the Gradium STT service.
+
+    Parameters:
+        delay_in_frames: Delay in audio frames (80ms each) before text is
+            generated. Higher delays allow more context but increase latency.
+    """
+
+    delay_in_frames: int = field(default_factory=lambda: NOT_GIVEN)
+
+
 class GradiumSTTService(WebsocketSTTService):
     """Gradium real-time speech-to-text service.
 
@@ -127,9 +141,15 @@ class GradiumSTTService(WebsocketSTTService):
         self._api_key = api_key
         self._api_endpoint_base_url = api_endpoint_base_url
         self._websocket = None
-        self._params = params or GradiumSTTService.InputParams()
         self._json_config = json_config
 
+        params = params or GradiumSTTService.InputParams()
+
+        self._settings: GradiumSTTSettings = GradiumSTTSettings(
+            language=params.language,
+            delay_in_frames=params.delay_in_frames if params.delay_in_frames else NOT_GIVEN,
+        )
+
         self._receive_task = None
 
         self._audio_buffer = bytearray()
@@ -149,16 +169,22 @@ class GradiumSTTService(WebsocketSTTService):
         """
         return True
 
-    async def set_language(self, language: Language):
-        """Set the recognition language and reconnect.
+    async def _update_settings_from_typed(self, update: STTSettings) -> set[str]:
+        """Apply a typed settings update, sync params, and reconnect.
 
         Args:
-            language: The language to use for speech recognition.
+            update: A :class:`STTSettings` (or ``GradiumSTTSettings``) delta.
+
+        Returns:
+            Set of field names whose values actually changed.
         """
-        logger.info(f"Switching STT language to: [{language}]")
-        self._params.language = language
+        changed = await super()._update_settings_from_typed(update)
+        if not changed:
+            return changed
+
         await self._disconnect()
         await self._connect()
+        return changed
 
     async def start(self, frame: StartFrame):
         """Start the speech-to-text service.
@@ -298,12 +324,12 @@ class GradiumSTTService(WebsocketSTTService):
             json_config = {}
             if self._json_config:
                 json_config = json.loads(self._json_config)
-            if self._params.language:
-                gradium_language = language_to_gradium_language(self._params.language)
+            if is_given(self._settings.language) and self._settings.language:
+                gradium_language = language_to_gradium_language(self._settings.language)
                 if gradium_language:
                     json_config["language"] = gradium_language
-            if self._params.delay_in_frames:
-                json_config["delay_in_frames"] = self._params.delay_in_frames
+            if is_given(self._settings.delay_in_frames) and self._settings.delay_in_frames:
+                json_config["delay_in_frames"] = self._settings.delay_in_frames
             if json_config:
                 setup_msg["json_config"] = json_config
             await self._websocket.send(json.dumps(setup_msg))