diff --git a/src/agents/voice/model.py b/src/agents/voice/model.py index 3b4a8e85b5..079704e56e 100644 --- a/src/agents/voice/model.py +++ b/src/agents/voice/model.py @@ -7,6 +7,7 @@ from typing_extensions import TypedDict +from .exceptions import UserError from .imports import np, npt from .input import AudioInput, StreamedAudioInput from .utils import get_sentence_based_splitter @@ -87,6 +88,15 @@ class TTSModelSettings: speed: float | None = None """The speed with which the TTS model will read the text. Between 0.25 and 4.0.""" + def __post_init__(self) -> None: + # Configurations loaded from JSON/YAML commonly represent NumPy dtypes as strings. + # Normalize those spellings once at the settings boundary so downstream consumers can + # compare against the supported NumPy dtypes consistently. + try: + self.dtype = np.dtype(self.dtype) + except (TypeError, ValueError) as error: + raise UserError("Invalid output dtype") from error + class TTSModel(abc.ABC): """A text-to-speech model that can convert text into audio output.""" diff --git a/tests/voice/test_tts_model_settings.py b/tests/voice/test_tts_model_settings.py new file mode 100644 index 0000000000..2b3ea77294 --- /dev/null +++ b/tests/voice/test_tts_model_settings.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +import numpy as np +import pytest + +from agents.exceptions import UserError +from agents.voice import AudioInput, TTSModelSettings, VoicePipeline + +from .helpers import extract_events +from .pipeline_test_models import QueuedSTTModel, QueuedVoiceWorkflow, ZeroPcmTTSModel + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("dtype", "expected_dtype"), + [("int16", np.int16), ("float32", np.float32), ("f4", np.float32)], + ids=["int16-string", "float32-string", "float32-alias"], +) +async def test_voicepipeline_accepts_string_tts_dtype_from_dictionary_config( + dtype: str, + expected_dtype: type[np.int16] | type[np.float32], +) -> None: + fake_stt = QueuedSTTModel(["first"]) + fake_tts = ZeroPcmTTSModel() + pipeline = VoicePipeline( + workflow=QueuedVoiceWorkflow([["out_1"]]), + stt_model=fake_stt, + tts_model=fake_tts, + config={"tts_settings": {"buffer_size": 1, "dtype": dtype}}, + ) + + result = await pipeline.run(AudioInput(buffer=np.zeros(2, dtype=np.int16))) + events, audio_chunks = await extract_events(result) + + assert events == ["turn_started", "audio", "turn_ended", "session_ended"] + decoded_audio = np.frombuffer(audio_chunks[0], dtype=expected_dtype) + assert decoded_audio.dtype == np.dtype(expected_dtype) + + +@pytest.mark.parametrize( + "dtype", + ["not-a-dtype", {"names": ["x"], "formats": []}], + ids=["unparseable-string", "malformed-structured-dtype"], +) +def test_tts_model_settings_preserves_user_error_for_invalid_dtype(dtype: object) -> None: + with pytest.raises(UserError, match="Invalid output dtype"): + TTSModelSettings(dtype=dtype) # type: ignore[arg-type]