fix(voice): accept string dtype settings

src/agents/voice/result.py

@@ -106,9 +106,14 @@ def _transform_audio_buffer(
 
         np_array = np.frombuffer(combined_buffer, dtype=np.int16)
 
-        if output_dtype == np.int16:
+        try:
+            normalized_output_dtype = np.dtype(output_dtype)
+        except (TypeError, ValueError) as error:
+            raise UserError("Invalid output dtype") from error
+
+        if normalized_output_dtype == np.dtype(np.int16):
             return np_array
-        elif output_dtype == np.float32:
+        elif normalized_output_dtype == np.dtype(np.float32):
             return (np_array.astype(np.float32) / 32767.0).reshape(-1, 1)
         else:
             raise UserError("Invalid output dtype")

tests/voice/test_pipeline.py

@@ -483,6 +483,36 @@ async def test_voicepipeline_normalizes_nested_dictionary_config() -> None:
     await fake_tts.verify_audio("out_1", audio_chunks[0])
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("dtype", "expected_dtype"),
+    [("int16", np.int16), ("float32", np.float32)],
+    ids=["int16-string", "float32-string"],
+)
+async def test_voicepipeline_accepts_string_tts_dtype_from_dictionary_config(
+    dtype: str,
+    expected_dtype: npt.DTypeLike,
+) -> None:
+    fake_stt = QueuedSTTModel(["first"])
+    fake_tts = ZeroPcmTTSModel()
+    pipeline = VoicePipeline(
+        workflow=QueuedVoiceWorkflow([["out_1"]]),
+        stt_model=fake_stt,
+        tts_model=fake_tts,
+        config={
+            "tracing_disabled": True,
+            "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.asyncio
 async def test_queued_stt_model_shares_static_and_streamed_transcription_queue() -> None:
     stt = QueuedSTTModel(["static", "streamed"])