feat(tts): integrate Async TTS engine into pipecat

dot-env.template

@@ -1,6 +1,10 @@
 # Anthropic
 ANTHROPIC_API_KEY=...
 
+# Async
+ASYNCAI_API_KEY=...
+ASYNCAI_VOICE_ID=...
+
 # AWS
 AWS_SECRET_ACCESS_KEY=...
 AWS_ACCESS_KEY_ID=...

examples/foundational/07aa-interruptible-asyncai-http.py

@@ -0,0 +1,110 @@
+import argparse
+import os
+
+import aiohttp
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.services.openai.stt import OpenAISTTService
+from pipecat.services.asyncai.tts import AsyncAIHttpTTSService
+from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams
+from pipecat.transports.services.daily import DailyParams
+
+load_dotenv(override=True)
+
+
+# We store functions so objects (e.g. SileroVADAnalyzer) don't get
+# instantiated. The function will be called when the desired transport gets
+# selected.
+transport_params = {
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(),
+    ),
+    "twilio": lambda: FastAPIWebsocketParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(),
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(),
+    ),
+}
+
+
+async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_sigint: bool):
+    logger.info(f"Starting bot")
+
+    # Create an HTTP session
+    async with aiohttp.ClientSession() as session:
+        stt = OpenAISTTService(api_key=os.getenv("OPENAI_API_KEY"))
+
+        tts = AsyncAIHttpTTSService(
+            api_key=os.getenv("ASYNCAI_API_KEY", ""),
+            voice_id=os.getenv("ASYNCAI_VOICE_ID", ""),
+            aiohttp_session=session,
+        )
+
+        llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
+
+        messages = [
+            {
+                "role": "system",
+                "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
+            },
+        ]
+
+        context = OpenAILLMContext(messages)
+        context_aggregator = llm.create_context_aggregator(context)
+
+        pipeline = Pipeline(
+            [
+                transport.input(),  # Transport user input
+                stt,
+                context_aggregator.user(),  # User responses
+                llm,  # LLM
+                tts,  # TTS
+                transport.output(),  # Transport bot output
+                context_aggregator.assistant(),  # Assistant spoken responses
+            ]
+        )
+
+        task = PipelineTask(
+            pipeline,
+            params=PipelineParams(
+                enable_metrics=True,
+                enable_usage_metrics=True,
+            ),
+        )
+
+        @transport.event_handler("on_client_connected")
+        async def on_client_connected(transport, client):
+            logger.info(f"Client connected")
+            # Kick off the conversation.
+            messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+            await task.queue_frames([context_aggregator.user().get_context_frame()])
+
+        @transport.event_handler("on_client_disconnected")
+        async def on_client_disconnected(transport, client):
+            logger.info(f"Client disconnected")
+            await task.cancel()
+
+        runner = PipelineRunner(handle_sigint=handle_sigint)
+
+        await runner.run(task)
+
+
+if __name__ == "__main__":
+    from pipecat.examples.run import main
+
+    main(run_example, transport_params=transport_params)

examples/foundational/07aa-interruptible-asyncai.py

@@ -0,0 +1,111 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import argparse
+import os
+
+from dotenv import load_dotenv
+from loguru import logger
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.services.openai.stt import OpenAISTTService
+from pipecat.services.asyncai.tts import AsyncAITTSService
+from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams
+from pipecat.transports.services.daily import DailyParams
+
+load_dotenv(override=True)
+
+
+# We store functions so objects (e.g. SileroVADAnalyzer) don't get
+# instantiated. The function will be called when the desired transport gets
+# selected.
+transport_params = {
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(),
+    ),
+    "twilio": lambda: FastAPIWebsocketParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(),
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+        vad_analyzer=SileroVADAnalyzer(),
+    ),
+}
+
+
+async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_sigint: bool):
+    logger.info(f"Starting bot")
+
+    stt = OpenAISTTService(api_key=os.getenv("OPENAI_API_KEY"))
+
+    tts = AsyncAITTSService(
+        api_key=os.getenv("ASYNCAI_API_KEY", ""),
+        voice_id=os.getenv("ASYNCAI_VOICE_ID", ""),
+    )
+
+    llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
+
+    messages = [
+        {
+            "role": "system",
+            "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
+        },
+    ]
+
+    context = OpenAILLMContext(messages)
+    context_aggregator = llm.create_context_aggregator(context)
+
+    pipeline = Pipeline(
+        [
+            transport.input(),  # Transport user input
+            stt,
+            context_aggregator.user(),  # User responses
+            llm,  # LLM
+            tts,  # TTS
+            transport.output(),  # Transport bot output
+            context_aggregator.assistant(),  # Assistant spoken responses
+        ]
+    )
+
+    task = PipelineTask(
+        pipeline,
+        params=PipelineParams(
+            enable_metrics=True,
+            enable_usage_metrics=True,
+        ),
+    )
+
+    @transport.event_handler("on_client_connected")
+    async def on_client_connected(transport, client):
+        logger.info(f"Client connected")
+        # Kick off the conversation.
+        messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+        await task.queue_frames([context_aggregator.user().get_context_frame()])
+
+    @transport.event_handler("on_client_disconnected")
+    async def on_client_disconnected(transport, client):
+        logger.info(f"Client disconnected")
+        await task.cancel()
+
+    runner = PipelineRunner(handle_sigint=handle_sigint)
+
+    await runner.run(task)
+
+
+if __name__ == "__main__":
+    from pipecat.examples.run import main
+
+    main(run_example, transport_params=transport_params)

src/pipecat/services/asyncai/__init__.py

@@ -0,0 +1,13 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import sys
+
+from pipecat.services import DeprecatedModuleProxy
+
+from .tts import *
+
+sys.modules[__name__] = DeprecatedModuleProxy(globals(), "asyncai", "asyncai.tts")

src/pipecat/services/asyncai/tts.py

@@ -0,0 +1,517 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Async text-to-speech service implementations."""
+
+import base64
+import json
+import uuid
+from typing import AsyncGenerator, Optional
+
+from loguru import logger
+from pydantic import BaseModel
+import asyncio
+import aiohttp
+
+from pipecat.frames.frames import (
+    CancelFrame,
+    EndFrame,
+    ErrorFrame,
+    Frame,
+    StartFrame,
+    StartInterruptionFrame,
+    TTSAudioRawFrame,
+    TTSStartedFrame,
+    TTSStoppedFrame,
+)
+from pipecat.processors.frame_processor import FrameDirection
+from pipecat.services.tts_service import AudioContextWordTTSService, TTSService
+from pipecat.transcriptions.language import Language
+from pipecat.utils.asyncio.watchdog_async_iterator import WatchdogAsyncIterator
+from pipecat.utils.tracing.service_decorators import traced_tts
+
+try:
+    import websockets
+except ModuleNotFoundError as e:
+    logger.error(f"Exception: {e}")
+    logger.error("In order to use Async, you need to `pip install pipecat-ai[asyncai]`.")
+    raise Exception(f"Missing module: {e}")
+
+
+def language_to_async_language(language: Language) -> Optional[str]:
+    """Convert a Language enum to Async language code.
+
+    Args:
+        language: The Language enum value to convert.
+
+    Returns:
+        The corresponding Async language code, or None if not supported.
+    """
+    BASE_LANGUAGES = {
+        Language.EN: "en",
+    }
+
+    result = BASE_LANGUAGES.get(language)
+
+    # If not found in base languages, try to find the base language from a variant
+    if not result:
+        # Convert enum value to string and get the base language part (e.g. en-En -> en)
+        lang_str = str(language.value)
+        base_code = lang_str.split("-")[0].lower()
+        # Look up the base code in our supported languages
+        result = base_code if base_code in BASE_LANGUAGES.values() else None
+
+    return result
+
+
+class AsyncAITTSService(AudioContextWordTTSService):
+    """Async TTS service with WebSocket streaming.
+
+    Provides text-to-speech using Async's streaming WebSocket API.
+    """
+
+    class InputParams(BaseModel):
+        """Input parameters for Async TTS configuration.
+
+        Parameters:
+            language: Language to use for synthesis.
+        """
+
+        language: Optional[Language] = Language.EN
+
+    def __init__(
+        self,
+        *,
+        api_key: str,
+        voice_id: str,
+        version: str = "v1",
+        url: str = "wss://api.async.ai/text_to_speech/websocket/ws",
+        model: str = "asyncflow_v2.0",
+        sample_rate: int = 32000,
+        encoding: str = "pcm_s16le",
+        container: str = "raw",
+        params: Optional[InputParams] = None,
+        aggregate_sentences: Optional[bool] = True,
+        **kwargs,
+    ):
+        """Initialize the Async TTS service.
+
+        Args:
+            api_key: Async API key.
+            voice_id: ID of the voice to use for synthesis.
+            version: Async API version.
+            url: WebSocket URL for Async TTS API.
+            model: TTS model to use (e.g., "asyncflow_v2.0").
+            sample_rate: Audio sample rate.
+            encoding: Audio encoding format.
+            container: Audio container format.
+            params: Additional input parameters for voice customization.
+            aggregate_sentences: Whether to aggregate sentences within the TTSService.
+            **kwargs: Additional arguments passed to the parent service.
+        """
+        # Aggregating sentences still gives cleaner-sounding results and fewer
+        # artifacts than streaming one word at a time. On average, waiting for a
+        # full sentence should only "cost" us 15ms or so with GPT-4o or a Llama
+        # 3 model, and it's worth it for the better audio quality.
+        #
+        # We also don't want to automatically push LLM response text frames,
+        # because the context aggregators will add them to the LLM context even
+        # if we're interrupted. 
+        super().__init__(
+            aggregate_sentences=aggregate_sentences,
+            push_text_frames=False,
+            pause_frame_processing=True,
+            sample_rate=sample_rate,
+            **kwargs,
+        )
+
+        params = params or AsyncAITTSService.InputParams()
+
+        self._api_key = api_key
+        self._api_version = version
+        self._url = url
+        self._settings = {
+            "output_format": {
+                "container": container,
+                "encoding": encoding,
+                "sample_rate": sample_rate,
+            },
+            "language": self.language_to_service_language(params.language)
+            if params.language
+            else "en",
+        }
+        
+        self.set_model_name(model)
+        self.set_voice(voice_id)
+        self._global_context_id = str(uuid.uuid4())
+
+        self._context_id = None
+        self._receive_task = None
+        self._keepalive_task = None
+
+    def can_generate_metrics(self) -> bool:
+        """Check if this service can generate processing metrics.
+
+        Returns:
+            True, as Async service supports metrics generation.
+        """
+        return True
+
+
+    def language_to_service_language(self, language: Language) -> Optional[str]:
+        """Convert a Language enum to Async language format.
+
+        Args:
+            language: The language to convert.
+
+        Returns:
+            The Async-specific language code, or None if not supported.
+        """
+        return language_to_async_language(language)
+
+    def _build_msg(
+        self, text: str = "", force: bool = False 
+    ):
+        msg = {
+            "transcript": text,
+            "force": force
+        }
+        return json.dumps(msg)
+    
+    async def start(self, frame: StartFrame):
+        """Start the Async TTS service.
+
+        Args:
+            frame: The start frame containing initialization parameters.
+        """
+        await super().start(frame)
+        await self._connect()
+
+    async def stop(self, frame: EndFrame):
+        """Stop the Async TTS service.
+
+        Args:
+            frame: The end frame.
+        """
+        await super().stop(frame)
+        await self._disconnect()
+
+    async def cancel(self, frame: CancelFrame):
+        """Cancel the Async TTS service.
+
+        Args:
+            frame: The cancel frame.
+        """
+        await super().cancel(frame)
+        await self._disconnect()
+
+    async def _connect(self):
+        await self._connect_websocket()    
+
+        if self._websocket and not self._receive_task:
+            self._receive_task = self.create_task(self._receive_task_handler(self._report_error))
+
+        if self._websocket and not self._keepalive_task:
+            self._keepalive_task = self.create_task(self._keepalive_task_handler())
+
+    async def _disconnect(self):
+        if self._receive_task:
+            await self.cancel_task(self._receive_task)
+            self._receive_task = None
+        
+        if self._keepalive_task:
+            await self.cancel_task(self._keepalive_task)
+            self._keepalive_task = None
+
+        await self._disconnect_websocket()
+
+    async def _connect_websocket(self):
+        try:
+            if self._websocket and self._websocket.open:
+                return
+            logger.debug("Connecting to Async")
+            self._websocket = await websockets.connect(
+                f"{self._url}?api_key={self._api_key}&version={self._api_version}"
+            )
+            init_msg = {
+                "model_id": self._model_name,
+                "voice": {"mode": "id", "id": self._voice_id},
+                "output_format": self._settings["output_format"],
+                "language": self._settings["language"]
+            }
+
+            await self._get_websocket().send(json.dumps(init_msg))
+        except Exception as e:
+            logger.error(f"{self} initialization error: {e}")
+            self._websocket = None
+            await self._call_event_handler("on_connection_error", f"{e}")
+
+    async def _disconnect_websocket(self):
+        try:
+            await self.stop_all_metrics()
+
+            if self._websocket:
+                logger.debug("Disconnecting from Async")
+                await self._websocket.close()
+        except Exception as e:
+            logger.error(f"{self} error closing websocket: {e}")
+        finally:
+            self._context_id = None
+            self._websocket = None
+
+    def _get_websocket(self):
+        if self._websocket:
+            return self._websocket
+        raise Exception("Websocket not connected")
+
+    async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection):
+        await super()._handle_interruption(frame, direction)
+        await self.stop_all_metrics()
+        if self._context_id:
+            self._context_id = None
+
+    async def flush_audio(self):
+        """Flush any pending audio and finalize the current context."""
+        if not self._context_id or not self._websocket:
+            return
+        logger.trace(f"{self}: flushing audio")
+        msg = self._build_msg(text=" ", force=True)
+        await self._websocket.send(msg)
+        self._context_id = None
+
+    async def _receive_messages(self):
+        async for message in WatchdogAsyncIterator(
+            self._get_websocket(), manager=self.task_manager
+        ):
+            msg = json.loads(message)
+            context_id = self._global_context_id
+            if not msg:
+                continue
+
+            if "final" in msg and msg["final"] is True:
+                await self.stop_ttfb_metrics()
+                await self.remove_audio_context(context_id)
+            elif msg.get("audio"):
+                await self.stop_ttfb_metrics()
+                frame = TTSAudioRawFrame(
+                    audio=base64.b64decode(msg["audio"]),
+                    sample_rate=self.sample_rate,
+                    num_channels=1,
+                )
+                await self.append_to_audio_context(context_id, frame)
+                
+            elif msg.get("error_code"):
+                logger.error(f"{self} error: {msg}")
+                await self.push_frame(TTSStoppedFrame())
+                await self.stop_all_metrics()
+                await self.push_error(ErrorFrame(f"{self} error: {msg['message']}"))
+                self._context_id = None
+            else:
+                logger.error(f"{self} error, unknown message type: {msg}")
+
+    async def _keepalive_task_handler(self):
+        """Send periodic keepalive messages to maintain WebSocket connection."""
+        KEEPALIVE_SLEEP = 10 if self.task_manager.task_watchdog_enabled else 3
+        while True:
+            self.reset_watchdog()
+            await asyncio.sleep(KEEPALIVE_SLEEP)
+            try:
+                if self._websocket and self._websocket.open:
+                    keepalive_message = {"transcript": " "}
+                    logger.trace("Sending keepalive message")
+                    await self._websocket.send(json.dumps(keepalive_message))
+            except websockets.ConnectionClosed as e:
+                logger.warning(f"{self} keepalive error: {e}")
+                break
+
+    @traced_tts
+    async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
+        """Generate speech from text using Async API websocket endpoint.
+
+        Args:
+            text: The text to synthesize into speech.
+
+        Yields:
+            Frame: Audio frames containing the synthesized speech.
+        """
+        logger.debug(f"{self}: Generating TTS [{text}]")
+
+        try:
+            if not self._websocket or self._websocket.closed:
+                await self._connect()
+
+            if not self._context_id:
+                await self.start_ttfb_metrics()
+                yield TTSStartedFrame()
+                self._context_id = self._global_context_id
+                await self.create_audio_context(self._context_id)
+
+            msg = self._build_msg(text=text)
+
+            try:
+                await self._get_websocket().send(msg)
+                await self.start_tts_usage_metrics(text)
+            except Exception as e:
+                logger.error(f"{self} error sending message: {e}")
+                yield TTSStoppedFrame()
+                await self._disconnect()
+                await self._connect()
+                return
+            yield None
+        except Exception as e:
+            logger.error(f"{self} exception: {e}")
+
+
+
+class AsyncAIHttpTTSService(TTSService):
+    """HTTP-based Async TTS service.
+
+    Provides text-to-speech using Asyncs' HTTP streaming API for simpler,
+    non-WebSocket integration. Suitable for use cases where streaming WebSocket
+    connection is not required or desired.
+    """
+
+    class InputParams(BaseModel):
+        """Input parameters for Async API.
+
+        Parameters:
+            language: Language to use for synthesis.
+        """
+        language: Optional[Language] = Language.EN
+    
+    def __init__(
+        self,
+        *,
+        api_key: str,
+        voice_id: str,
+        aiohttp_session: aiohttp.ClientSession,
+        model: str = "asyncflow_v2.0",
+        url: str = "https://api.async.ai",
+        version: str = "v1",
+        sample_rate: int = 32000,
+        encoding: str = "pcm_s16le",
+        container: str = "raw",
+        params: Optional[InputParams] = None,
+        **kwargs,
+    ):
+        """Initialize the Async TTS service.
+
+        Args:
+            api_key: Async API key.
+            voice_id: ID of the voice to use for synthesis.
+            model: TTS model to use (e.g., "asyncflow_v2.0").
+            url: Base URL for Async API.
+            version: API version string for Async API.
+            sample_rate: Audio sample rate.
+            encoding: Audio encoding format.
+            container: Audio container format.
+            params: Additional input parameters for voice customization.
+            **kwargs: Additional arguments passed to the parent TTSService.
+        """
+        super().__init__(sample_rate=sample_rate, **kwargs)
+
+        params = params or AsyncAIHttpTTSService.InputParams()
+
+        self._api_key = api_key
+        self._base_url = url
+        self._api_version = version
+        self._settings = {
+            "output_format": {
+                "container": container,
+                "encoding": encoding,
+                "sample_rate": sample_rate,
+            },
+            "language": self.language_to_service_language(params.language)
+            if params.language
+            else "en",
+        }
+        self.set_voice(voice_id)
+        self.set_model_name(model)
+
+        self._session = aiohttp_session
+
+    def can_generate_metrics(self) -> bool:
+        """Check if this service can generate processing metrics.
+
+        Returns:
+            True, as Async HTTP service supports metrics generation.
+        """
+        return True
+
+    def language_to_service_language(self, language: Language) -> Optional[str]:
+        """Convert a Language enum to Async language format.
+
+        Args:
+            language: The language to convert.
+
+        Returns:
+            The Async-specific language code, or None if not supported.
+        """
+        return language_to_async_language(language)
+
+    async def start(self, frame: StartFrame):
+        """Start the Async HTTP TTS service.
+
+        Args:
+            frame: The start frame containing initialization parameters.
+        """
+        await super().start(frame)
+
+    @traced_tts
+    async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
+        """Generate speech from text using Asyncs' 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()
+                    logger.error(f"Async API error: {error_text}")
+                    await self.push_error(ErrorFrame(f"Async API error: {error_text}"))
+                    raise Exception(f"Async API returned status {response.status}: {error_text}")
+
+                audio_data = await response.read()
+
+            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:
+            logger.error(f"{self} exception: {e}")
+            await self.push_error(ErrorFrame(f"Error generating TTS: {e}"))
+        finally:
+            await self.stop_ttfb_metrics()
+            yield TTSStoppedFrame()