fix: silence _settings NotGiven leaks and tighten Google STT language method

Six pyright errors followed the same pattern: a value flowed out of
`self._settings.X` (typed `T | _NotGiven`) into a context that wanted
the plain `T`. Wrap each with `assert_given(...)` so the sentinel
gets stripped at the boundary:

- aws/nova_sonic/llm.py: `_settings.model` (in InvokeModel...Input)
  and `_settings.system_instruction` (passed to the adapter).
- deepgram/flux/base.py: iterating `_settings.keyterm`.
- google/stt.py: iterating `_settings.languages`.
- google/tts.py: iterating `_settings.speaker_configs`.
- openai/base_llm.py: `_settings.system_instruction` passed to the
  adapter.

Also takes a deeper pass at the related Google STT issue: the override
of `language_to_service_language` had been broadened to take
`Language | list[Language]` and return `str | list[str]`, a Liskov
violation against the base's `Language -> str | None` contract.
External callers always pass a single Language, and the only consumer
of the list path was Google STT's own `_get_language_codes`. Restore
the override to a single-Language signature and let
`_get_language_codes` iterate. The override is also tightened to
return `str` (narrower than the base's `str | None`, which is
LSP-compatible) since it always falls back to `"en-US"` rather than
returning None.

Net: -7 pyright errors (full-config run: 782 -> 775).
This commit is contained in:
Paul Kompfner
2026-04-28 16:19:14 -04:00
parent 2a731336be
commit ef226c8a8e
5 changed files with 20 additions and 14 deletions

View File

@@ -974,7 +974,9 @@ class AWSNovaSonicLLMService(LLMService[AWSNovaSonicLLMAdapter]):
async def open_stream(self, client): async def open_stream(self, client):
"""Open a bidirectional stream on the given client.""" """Open a bidirectional stream on the given client."""
return await client.invoke_model_with_bidirectional_stream( return await client.invoke_model_with_bidirectional_stream(
InvokeModelWithBidirectionalStreamOperationInput(model_id=self._settings.model) InvokeModelWithBidirectionalStreamOperationInput(
model_id=assert_given(self._settings.model)
)
) )
async def send_event(self, event_json: str, stream): async def send_event(self, event_json: str, stream):
@@ -1127,7 +1129,7 @@ class AWSNovaSonicLLMService(LLMService[AWSNovaSonicLLMAdapter]):
return None, [] return None, []
adapter = self.get_llm_adapter() adapter = self.get_llm_adapter()
llm_params = adapter.get_llm_invocation_params( llm_params = adapter.get_llm_invocation_params(
self._context, system_instruction=self._settings.system_instruction self._context, system_instruction=assert_given(self._settings.system_instruction)
) )
tools = ( tools = (
llm_params["tools"] llm_params["tools"]

View File

@@ -253,7 +253,7 @@ class DeepgramFluxSTTBase(STTService):
params.append(f"mip_opt_out={str(self._mip_opt_out).lower()}") params.append(f"mip_opt_out={str(self._mip_opt_out).lower()}")
# Add keyterm parameters (can have multiple) # Add keyterm parameters (can have multiple)
for keyterm in self._settings.keyterm: for keyterm in assert_given(self._settings.keyterm):
params.append(urlencode({"keyterm": keyterm})) params.append(urlencode({"keyterm": keyterm}))
# Add tag parameters (can have multiple) # Add tag parameters (can have multiple)

View File

@@ -620,17 +620,20 @@ class GoogleSTTService(STTService):
""" """
return True return True
def language_to_service_language(self, language: Language | list[Language]) -> str | list[str]: def language_to_service_language(self, language: Language) -> str:
"""Convert Language enum(s) to Google STT language code(s). """Convert a Language enum to a Google STT language code.
Narrower return type than the base class's ``str | None``: this
override always returns a string, falling back to ``"en-US"`` for
languages not in the verified mapping (see
:func:`language_to_google_stt_language`).
Args: Args:
language: Single Language enum or list of Language enums. language: The Language enum value to convert.
Returns: Returns:
str | List[str]: Google STT language code(s). The Google STT language code.
""" """
if isinstance(language, list):
return [language_to_google_stt_language(lang) or "en-US" for lang in language]
return language_to_google_stt_language(language) or "en-US" return language_to_google_stt_language(language) or "en-US"
def _get_language_codes(self) -> list[str]: def _get_language_codes(self) -> list[str]:
@@ -642,8 +645,9 @@ class GoogleSTTService(STTService):
Returns: Returns:
List[str]: Google STT language code strings. List[str]: Google STT language code strings.
""" """
if self._settings.languages: languages = assert_given(self._settings.languages)
return [self.language_to_service_language(lang) for lang in self._settings.languages] if languages:
return [self.language_to_service_language(lang) for lang in languages]
language_codes = assert_given(self._settings.language_codes) language_codes = assert_given(self._settings.language_codes)
if language_codes: if language_codes:
return list(language_codes) return list(language_codes)

View File

@@ -1417,7 +1417,7 @@ class GeminiTTSService(GoogleBaseTTSService):
if self._settings.multi_speaker and self._settings.speaker_configs: if self._settings.multi_speaker and self._settings.speaker_configs:
# Multi-speaker mode # Multi-speaker mode
speaker_voice_configs = [] speaker_voice_configs = []
for speaker_config in self._settings.speaker_configs: for speaker_config in assert_given(self._settings.speaker_configs):
speaker_voice_configs.append( speaker_voice_configs.append(
texttospeech_v1.MultispeakerPrebuiltVoice( texttospeech_v1.MultispeakerPrebuiltVoice(
speaker_alias=speaker_config["speaker_alias"], speaker_alias=speaker_config["speaker_alias"],

View File

@@ -39,7 +39,7 @@ from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.frame_processor import FrameDirection from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.llm_service import FunctionCallFromLLM, LLMService from pipecat.services.llm_service import FunctionCallFromLLM, LLMService
from pipecat.services.settings import NOT_GIVEN as _NOT_GIVEN from pipecat.services.settings import NOT_GIVEN as _NOT_GIVEN
from pipecat.services.settings import LLMSettings, _NotGiven from pipecat.services.settings import LLMSettings, _NotGiven, assert_given
from pipecat.utils.tracing.service_decorators import traced_llm from pipecat.utils.tracing.service_decorators import traced_llm
@@ -299,7 +299,7 @@ class BaseOpenAILLMService(LLMService[OpenAILLMAdapter]):
params_from_context = adapter.get_llm_invocation_params( params_from_context = adapter.get_llm_invocation_params(
context, context,
system_instruction=self._settings.system_instruction, system_instruction=assert_given(self._settings.system_instruction),
convert_developer_to_user=not self.supports_developer_role, convert_developer_to_user=not self.supports_developer_role,
) )