fix(voice): accept string TTS dtypes

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,17 @@ 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:
+        # Serialized configs commonly carry NumPy dtypes as strings. Normalize only those string
+        # spellings: custom TTS providers receive this settings object directly and may rely on
+        # non-string DTypeLike inputs (for example the callable np.int16 scalar class) retaining
+        # their original representation.
+        if isinstance(self.dtype, str):
+            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."""

tests/voice/test_tts_model_settings.py

@@ -0,0 +1,77 @@
+from __future__ import annotations
+
+from collections.abc import AsyncIterator
+
+import numpy as np
+import pytest
+
+from agents.exceptions import UserError
+from agents.voice import AudioInput, TTSModelSettings, VoicePipeline
+
+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: list[str] = []
+    seen_dtypes: list[np.dtype[object]] = []
+    async for event in result.stream():
+        if event.type == "voice_stream_event_audio":
+            events.append("audio")
+            if event.data is not None:
+                seen_dtypes.append(event.data.dtype)
+        elif event.type == "voice_stream_event_lifecycle":
+            events.append(event.event)
+        elif event.type == "voice_stream_event_error":
+            events.append("error")
+
+    assert events == ["turn_started", "audio", "turn_ended", "session_ended"]
+    assert seen_dtypes
+    assert all(dtype == np.dtype(expected_dtype) for dtype in seen_dtypes)
+
+
+@pytest.mark.parametrize("dtype", ["not-a-dtype"], ids=["unparseable-string"])
+def test_tts_model_settings_preserves_user_error_for_invalid_string_dtype(dtype: str) -> None:
+    with pytest.raises(UserError, match="Invalid output dtype"):
+        TTSModelSettings(dtype=dtype)
+
+
+@pytest.mark.asyncio
+async def test_voicepipeline_preserves_non_string_dtype_for_custom_tts_provider() -> None:
+    class CallableDtypeTTSModel(ZeroPcmTTSModel):
+        async def run(self, text: str, settings: TTSModelSettings) -> AsyncIterator[bytes]:
+            del text
+            # Custom providers historically receive the original DTypeLike object. In particular,
+            # NumPy scalar classes are callable and provider code may legitimately use that API.
+            assert settings.dtype is np.int16
+            assert settings.dtype(1) == np.int16(1)  # type: ignore[operator]
+            yield np.zeros(2, dtype=np.int16).tobytes()
+
+    pipeline = VoicePipeline(
+        workflow=QueuedVoiceWorkflow([["out_1"]]),
+        stt_model=QueuedSTTModel(["first"]),
+        tts_model=CallableDtypeTTSModel(),
+        config={"tts_settings": {"buffer_size": 1, "dtype": np.int16}},
+    )
+
+    result = await pipeline.run(AudioInput(buffer=np.zeros(2, dtype=np.int16)))
+    async for _ in result.stream():
+        pass