Repository: livekit/agents
File: livekit-plugins/livekit-plugins-deepgram/livekit/plugins/deepgram/tts.py (Python)

Code before:
                )

                async for data, _ in resp.content.iter_chunks():
                    output_emitter.push(data)

                output_emitter.flush()

        except asyncio.TimeoutError:
            raise APITimeoutError() from None
        except aiohttp.ClientResponseError as e:
            raise APIStatusError(
                message=e.message, status_code=e.status, request_id=None, body=None
            ) from None
        except Exception as e:
            raise APIConnectionError() from e


class SynthesizeStream(tts.SynthesizeStream):
    _CLOSE_MSG: str = json.dumps({"type": "CloseStream"})

    def __init__(self, *, tts: TTS, conn_options: APIConnectOptions):
        super().__init__(tts=tts, conn_options=conn_options)
        self._tts: TTS = tts
        self._opts = replace(tts._opts)
        self._segments_ch = utils.aio.Chan[tokenize.WordStream]()

    async def _run(self, output_emitter: tts.AudioEmitter) -> None:
        request_id = utils.shortuuid()
        output_emitter.initialize(
            request_id=request_id,
            sample_rate=self._opts.sample_rate,
            num_channels=1,
            mime_type="audio/pcm",
            stream=True,
        )

        async def _tokenize_input() -> None:
            # Converts incoming text into WordStreams and sends them into _segments_ch
            word_stream = None
            async for input in self._input_ch:
                if isinstance(input, str):
                    if word_stream is None:
                        word_stream = self._opts.word_tokenizer.stream()
                        self._segments_ch.send_nowait(word_stream)
                    word_stream.push_text(input)
                elif isinstance(input, self._FlushSentinel):
                    if word_stream:
                        word_stream.end_input()
                    word_stream = None

            self._segments_ch.close()

Review comment (bug):
Few Suggestions:

1) Correct control message (TTS):
`await ws.send_str(json.dumps({"type": "Close"}))  # not "CloseStream"`

2) Flush before close (if any text was streamed):
```
await ws.send_str(json.dumps({"type": "Flush"}))
await ws.send_str(json.dumps({"type": "Close"}))
```

3) Avoid fixed sleep; wait briefly for server ack:
```
try:
    msg = await asyncio.wait_for(ws.receive(), timeout=1.0)
except asyncio.TimeoutError:
    pass
```

4) Drain one message, then close (prevents race conditions):
```
try:
    msg = await asyncio.wait_for(ws.receive(), timeout=1.0)
    # optional: print(msg.type, msg.data)
except asyncio.TimeoutError:
    pass
finally:
    await ws.close()
```

These set of  tweaks should eliminate lingering TTS sessions and prevent projects from hitting concurrency caps (429 Too Many Requests).

Also, here is  a minimal example using the Deepgram Python SDK that demonstrates the proper “Flush + Close” sequence for TTS: https://github.com/deepgram/deepgram-python-sdk/blob/main/examples/speak/v1/connect/async.py

Code after:
                output_emitter.initialize(
                    request_id=utils.shortuuid(),
                    sample_rate=self._opts.sample_rate,
                    num_channels=NUM_CHANNELS,
                    mime_type="audio/pcm",
                )

                async for data, _ in resp.content.iter_chunks():
                    output_emitter.push(data)

                output_emitter.flush()

        except asyncio.TimeoutError:
            raise APITimeoutError() from None
        except aiohttp.ClientResponseError as e:
            raise APIStatusError(
                message=e.message, status_code=e.status, request_id=None, body=None
            ) from None
        except Exception as e:
            raise APIConnectionError() from e


class SynthesizeStream(tts.SynthesizeStream):
    _FLUSH_MSG: str = json.dumps({"type": "Flush"})
    _CLOSE_MSG: str = json.dumps({"type": "Close"})

    def __init__(self, *, tts: TTS, conn_options: APIConnectOptions):
        super().__init__(tts=tts, conn_options=conn_options)
        self._tts: TTS = tts
        self._opts = replace(tts._opts)
        self._segments_ch = utils.aio.Chan[tokenize.WordStream]()

    async def _run(self, output_emitter: tts.AudioEmitter) -> None:
        request_id = utils.shortuuid()
        output_emitter.initialize(
            request_id=request_id,
            sample_rate=self._opts.sample_rate,
            num_channels=1,
            mime_type="audio/pcm",
            stream=True,
        )

        async def _tokenize_input() -> None:
            # Converts incoming text into WordStreams and sends them into _segments_ch
            word_stream = None
            async for input in self._input_ch:
                if isinstance(input, str):
                    if word_stream is None:
                        word_stream = self._opts.word_tokenizer.stream()
                        self._segments_ch.send_nowait(word_stream)
                    word_stream.push_text(input)