Fix TTFB metric and add multi-context WebSocket support for Async TTS

[ashotbagh]

Summary

This PR improves the Async TTS integration by fixing TTFB metric calculation and making WebSocket-based interruption handling more robust.

Changes

• Fixed incorrect TTFB metric calculation in AsyncAIHttpTTSService.
• Updated AsyncAITTSService to support multi-context WebSocket sessions, improving reliability when handling interruptions and rapid context switches.

Design Notes

In tts_service.py, the only abstract class that provides built-in audio context management is AudioContextWordTTSService, which assumes the availability of word-level timestamps.

Since Async does not currently provide word timestamps, the Async TTS implementation cannot derive from this class. Instead, it derives from WebsocketTTSService, and the required context-management logic is implemented directly in the Async service.

It may be beneficial to introduce a base class for audio context management that does not depend on word timestamps, which would allow services like Async to reuse this logic more cleanly.

[codecov]

Codecov Report

❌ Patch coverage is 0% with 63 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
src/pipecat/services/asyncai/tts.py	0.00%	63 Missing ⚠️

Files with missing lines	Coverage Δ
src/pipecat/services/asyncai/tts.py	0.00% <0.00%> (ø)

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[markbackman]

FYI: I'm proposing an AudioContextTTSService base class here: #3289. If we decide to include that, it will greatly simplify changes to this class.

[markbackman]

We now use changelog fragements. You can learn how to provide changelog updates here:
https://github.com/pipecat-ai/pipecat/blob/main/CONTRIBUTING.md#changelog-entries

[ashotbagh]

I’ve pushed an update adapting the Async TTS implementation to use the newly merged AudioContextTTSService. Thanks a lot for introducing this base class — it’s a great addition and it significantly simplifies context handling for services like Async.
One small question regarding the inheritance: since AudioContextTTSService is intended to serve as a TTS base, would it make sense for it to inherit from WebsocketTTSService (or otherwise include TTSService) so that downstream implementations automatically get the standard TTS helpers, metrics, and behavior?
For Async, I can adapt either way, but at the moment I still rely on several utilities that currently live in WebsocketTTSService / TTSService. Just wanted to flag this in case it’s useful for future refinements.

[markbackman]

One small question regarding the inheritance: since AudioContextTTSService is intended to serve as a TTS base, would it make sense for it to inherit from WebsocketTTSService (or otherwise include TTSService) so that downstream implementations automatically get the standard TTS helpers, metrics, and behavior?

You're right. This was an oversight. I'm making the suggested change here:
#3363

Hopefully this will be reviewed and on main at some point today.

[markbackman]

Ok, merged! If you update this PR, I'll be ready to review and test.

[ashotbagh]

Thanks a lot for the very fast fix and follow-up - much appreciated.

I’ve rebased on the latest main and updated the Async TTS implementation to align with the updated AudioContextTTSService.

Happy to adjust further if needed.

[markbackman]

This is the review for AsyncAITTSService. In addition, a few things:

• Make sure to lint the code. uv run scripts/fix-ruff.sh or install pre-commit hook (uv run pre-commit install)
• You'll want to rebase to get the latest changes.

[markbackman]

For the TTFB change, there's a more minimal change that doesn't require tracking first_byte_seen. Since it's an HTTP request, each request has a single response.

So, in your run_tts(), you can just do:

    @traced_tts
    async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
        """Generate speech from text using Async's HTTP streaming API.

        Args:
            text: The text to synthesize into speech.

        Yields:
            Frame: Audio frames containing the synthesized speech.
        """
        logger.debug(f"{self}: Generating TTS [{text}]")

        try:
            voice_config = {"mode": "id", "id": self._voice_id}
            await self.start_ttfb_metrics()
            payload = {
                "model_id": self._model_name,
                "transcript": text,
                "voice": voice_config,
                "output_format": self._settings["output_format"],
                "language": self._settings["language"],
            }
            yield TTSStartedFrame()
            headers = {
                "version": self._api_version,
                "x-api-key": self._api_key,
                "Content-Type": "application/json",
            }
            url = f"{self._base_url}/text_to_speech/streaming"

            async with self._session.post(url, json=payload, headers=headers) as response:
                if response.status != 200:
                    error_text = await response.text()
                    await self.push_error(error_msg=f"Async API error: {error_text}")
                    raise Exception(f"Async API returned status {response.status}: {error_text}")

                # Read streaming bytes; stop TTFB on the *first* received chunk
                buffer = bytearray()
                async for chunk in response.content.iter_chunked(64 * 1024):
                    if not chunk:
                        continue
                    await self.stop_ttfb_metrics()
                    buffer.extend(chunk)
                audio_data = bytes(buffer)

            await self.start_tts_usage_metrics(text)

            frame = TTSAudioRawFrame(
                audio=audio_data,
                sample_rate=self.sample_rate,
                num_channels=1,
            )

            yield frame

        except Exception as e:
            await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)
        finally:
            await self.stop_ttfb_metrics()
            yield TTSStoppedFrame()

Note: the only change is:

-                audio_data = await response.read()
+                # Read streaming bytes; stop TTFB on the *first* received chunk
+                buffer = bytearray()
+                async for chunk in response.content.iter_chunked(64 * 1024):
+                    if not chunk:
+                        continue
+                    await self.stop_ttfb_metrics()
+                    buffer.extend(chunk)
+                audio_data = bytes(buffer)

With this, I see sane TTFB values and correct character usage data.

[markbackman]

Also, we just merged this PR:
#3392

To get the benefits of it, for AsyncAITTSService, you should:

• In _connect() call await super()._connect() first
• In _disconnect(), call await super()._disconnect() first

This will add better websocket handling in the event of exceptions. It will handle reconnects, etc.

[ashotbagh]

Hi @markbackman,

All comments resolved, rebased on main, and linting is clean.
Ready for re-review. Thanks!

[markbackman]

LGTM! Nice clean up 🙌