From f2e9562f1b31f70f043b3617c3395616913c20c9 Mon Sep 17 00:00:00 2001 From: Ashot Date: Wed, 23 Jul 2025 14:44:21 +0400 Subject: [PATCH 1/4] feat(tts): integrate Async TTS engine into pipecat --- dot-env.template | 4 + .../07aa-interruptible-asyncai-http.py | 110 ++++ .../07aa-interruptible-asyncai.py | 111 ++++ src/pipecat/services/asyncai/__init__.py | 13 + src/pipecat/services/asyncai/tts.py | 517 ++++++++++++++++++ 5 files changed, 755 insertions(+) create mode 100644 examples/foundational/07aa-interruptible-asyncai-http.py create mode 100644 examples/foundational/07aa-interruptible-asyncai.py create mode 100644 src/pipecat/services/asyncai/__init__.py create mode 100644 src/pipecat/services/asyncai/tts.py diff --git a/dot-env.template b/dot-env.template index dbed59da8..4f96b8d73 100644 --- a/dot-env.template +++ b/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=... diff --git a/examples/foundational/07aa-interruptible-asyncai-http.py b/examples/foundational/07aa-interruptible-asyncai-http.py new file mode 100644 index 000000000..03ad158e5 --- /dev/null +++ b/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) diff --git a/examples/foundational/07aa-interruptible-asyncai.py b/examples/foundational/07aa-interruptible-asyncai.py new file mode 100644 index 000000000..49479999a --- /dev/null +++ b/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) diff --git a/src/pipecat/services/asyncai/__init__.py b/src/pipecat/services/asyncai/__init__.py new file mode 100644 index 000000000..9692138e6 --- /dev/null +++ b/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") diff --git a/src/pipecat/services/asyncai/tts.py b/src/pipecat/services/asyncai/tts.py new file mode 100644 index 000000000..7d2ace265 --- /dev/null +++ b/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() From a13b9544159db5c30342c8592256c5c22e678c83 Mon Sep 17 00:00:00 2001 From: Ashot Date: Sat, 26 Jul 2025 09:39:50 +0400 Subject: [PATCH 2/4] formatting/cleanup: address Copilot PR review comments --- .../07aa-interruptible-asyncai-http.py | 6 ++++++ src/pipecat/services/asyncai/tts.py | 18 +++++++----------- 2 files changed, 13 insertions(+), 11 deletions(-) diff --git a/examples/foundational/07aa-interruptible-asyncai-http.py b/examples/foundational/07aa-interruptible-asyncai-http.py index 03ad158e5..415976d61 100644 --- a/examples/foundational/07aa-interruptible-asyncai-http.py +++ b/examples/foundational/07aa-interruptible-asyncai-http.py @@ -1,3 +1,9 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + import argparse import os diff --git a/src/pipecat/services/asyncai/tts.py b/src/pipecat/services/asyncai/tts.py index 7d2ace265..aa90280ca 100644 --- a/src/pipecat/services/asyncai/tts.py +++ b/src/pipecat/services/asyncai/tts.py @@ -119,7 +119,7 @@ class AsyncAITTSService(AudioContextWordTTSService): # # 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. + # if we're interrupted. super().__init__( aggregate_sentences=aggregate_sentences, push_text_frames=False, @@ -142,7 +142,7 @@ class AsyncAITTSService(AudioContextWordTTSService): "language": self.language_to_service_language(params.language) if params.language else "en", - } + }, self.set_model_name(model) self.set_voice(voice_id) @@ -160,7 +160,6 @@ class AsyncAITTSService(AudioContextWordTTSService): """ return True - def language_to_service_language(self, language: Language) -> Optional[str]: """Convert a Language enum to Async language format. @@ -173,8 +172,8 @@ class AsyncAITTSService(AudioContextWordTTSService): return language_to_async_language(language) def _build_msg( - self, text: str = "", force: bool = False - ): + self, text: str = "", force: bool = False + ) -> str: msg = { "transcript": text, "force": force @@ -240,7 +239,7 @@ class AsyncAITTSService(AudioContextWordTTSService): "model_id": self._model_name, "voice": {"mode": "id", "id": self._voice_id}, "output_format": self._settings["output_format"], - "language": self._settings["language"] + "language": self._settings["language"], } await self._get_websocket().send(json.dumps(init_msg)) @@ -302,7 +301,6 @@ class AsyncAITTSService(AudioContextWordTTSService): 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()) @@ -364,12 +362,10 @@ class AsyncAITTSService(AudioContextWordTTSService): 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, + Provides text-to-speech using Async's HTTP streaming API for simpler, non-WebSocket integration. Suitable for use cases where streaming WebSocket connection is not required or desired. """ @@ -462,7 +458,7 @@ class AsyncAIHttpTTSService(TTSService): @traced_tts async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - """Generate speech from text using Asyncs' HTTP streaming API. + """Generate speech from text using Async's HTTP streaming API. Args: text: The text to synthesize into speech. From 83b4747196ae87b929608041a55b66fa893cfd5e Mon Sep 17 00:00:00 2001 From: Ashot Date: Mon, 28 Jul 2025 17:39:23 +0400 Subject: [PATCH 3/4] chore: address review comments --- CHANGELOG.md | 7 ++ docs/api/requirements.txt | 1 + ....py => 07ac-interruptible-asyncai-http.py} | 0 ...yncai.py => 07ac-interruptible-asyncai.py} | 0 pyproject.toml | 1 + src/pipecat/services/asyncai/__init__.py | 6 -- src/pipecat/services/asyncai/tts.py | 76 +++++++++---------- 7 files changed, 44 insertions(+), 47 deletions(-) rename examples/foundational/{07aa-interruptible-asyncai-http.py => 07ac-interruptible-asyncai-http.py} (100%) rename examples/foundational/{07aa-interruptible-asyncai.py => 07ac-interruptible-asyncai.py} (100%) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6d1b1f1a6..18c433fe3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- **async.ai TTS integration** (https://async.ai/) + - `AsyncAITTSService` – streaming / interruptible TTS over WebSocket. + - `AsyncAIHttpTTSService` – streaming TTS over HTTP. + - Example scripts: + - `examples/foundational/07ac-interruptible-asyncai.py` (WebSocket demo) + - `examples/foundational/07ac-interruptible-asyncai-http.py` (HTTP demo) + - Added a new TTS service, `InworldTTSService`. This service provides low-latency, high-quality speech generation using Inworld's streaming API. diff --git a/docs/api/requirements.txt b/docs/api/requirements.txt index b90b48f11..95e8bc451 100644 --- a/docs/api/requirements.txt +++ b/docs/api/requirements.txt @@ -8,6 +8,7 @@ toml # Install all extras individually to ensure they're properly resolved pipecat-ai[anthropic] pipecat-ai[assemblyai] +pipecat-ai[asyncai] pipecat-ai[aws] pipecat-ai[azure] pipecat-ai[cartesia] diff --git a/examples/foundational/07aa-interruptible-asyncai-http.py b/examples/foundational/07ac-interruptible-asyncai-http.py similarity index 100% rename from examples/foundational/07aa-interruptible-asyncai-http.py rename to examples/foundational/07ac-interruptible-asyncai-http.py diff --git a/examples/foundational/07aa-interruptible-asyncai.py b/examples/foundational/07ac-interruptible-asyncai.py similarity index 100% rename from examples/foundational/07aa-interruptible-asyncai.py rename to examples/foundational/07ac-interruptible-asyncai.py diff --git a/pyproject.toml b/pyproject.toml index 5ef4ddeef..8c57bc8d0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -45,6 +45,7 @@ Website = "https://pipecat.ai" [project.optional-dependencies] anthropic = [ "anthropic~=0.49.0" ] assemblyai = [ "websockets>=13.1,<15.0" ] +asyncai = [ "websockets>=13.1,<15.0" ] aws = [ "aioboto3~=15.0.0", "websockets>=13.1,<15.0" ] aws-nova-sonic = [ "aws_sdk_bedrock_runtime~=0.0.2; python_version>='3.12'" ] azure = [ "azure-cognitiveservices-speech~=1.42.0"] diff --git a/src/pipecat/services/asyncai/__init__.py b/src/pipecat/services/asyncai/__init__.py index 9692138e6..9542e4cd6 100644 --- a/src/pipecat/services/asyncai/__init__.py +++ b/src/pipecat/services/asyncai/__init__.py @@ -1,9 +1,3 @@ -# -# Copyright (c) 2024–2025, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - import sys from pipecat.services import DeprecatedModuleProxy diff --git a/src/pipecat/services/asyncai/tts.py b/src/pipecat/services/asyncai/tts.py index aa90280ca..d7d3dce00 100644 --- a/src/pipecat/services/asyncai/tts.py +++ b/src/pipecat/services/asyncai/tts.py @@ -28,13 +28,15 @@ from pipecat.frames.frames import ( TTSStoppedFrame, ) from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.tts_service import AudioContextWordTTSService, TTSService +from pipecat.services.tts_service import InterruptibleTTSService, 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 + from websockets.asyncio.client import connect as websocket_connect + from websockets.protocol import State except ModuleNotFoundError as e: logger.error(f"Exception: {e}") logger.error("In order to use Async, you need to `pip install pipecat-ai[asyncai]`.") @@ -67,7 +69,7 @@ def language_to_async_language(language: Language) -> Optional[str]: return result -class AsyncAITTSService(AudioContextWordTTSService): +class AsyncAITTSService(InterruptibleTTSService): """Async TTS service with WebSocket streaming. Provides text-to-speech using Async's streaming WebSocket API. @@ -90,7 +92,7 @@ class AsyncAITTSService(AudioContextWordTTSService): version: str = "v1", url: str = "wss://api.async.ai/text_to_speech/websocket/ws", model: str = "asyncflow_v2.0", - sample_rate: int = 32000, + sample_rate: Optional[int] = None, encoding: str = "pcm_s16le", container: str = "raw", params: Optional[InputParams] = None, @@ -112,18 +114,11 @@ class AsyncAITTSService(AudioContextWordTTSService): 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, + push_stop_frames=True, sample_rate=sample_rate, **kwargs, ) @@ -137,20 +132,19 @@ class AsyncAITTSService(AudioContextWordTTSService): "output_format": { "container": container, "encoding": encoding, - "sample_rate": sample_rate, + "sample_rate": 0, }, "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 + self._started = False def can_generate_metrics(self) -> bool: """Check if this service can generate processing metrics. @@ -187,6 +181,7 @@ class AsyncAITTSService(AudioContextWordTTSService): frame: The start frame containing initialization parameters. """ await super().start(frame) + self._settings["output_format"]["sample_rate"] = self.sample_rate await self._connect() async def stop(self, frame: EndFrame): @@ -229,10 +224,10 @@ class AsyncAITTSService(AudioContextWordTTSService): async def _connect_websocket(self): try: - if self._websocket and self._websocket.open: + if self._websocket and self._websocket.state is State.OPEN: return logger.debug("Connecting to Async") - self._websocket = await websockets.connect( + self._websocket = await websocket_connect( f"{self._url}?api_key={self._api_key}&version={self._api_version}" ) init_msg = { @@ -258,41 +253,41 @@ class AsyncAITTSService(AudioContextWordTTSService): except Exception as e: logger.error(f"{self} error closing websocket: {e}") finally: - self._context_id = None self._websocket = None + self._started = False 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: + """Flush any pending audio.""" + if 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 push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM): + """Push a frame downstream with special handling for stop conditions. + + Args: + frame: The frame to push. + direction: The direction to push the frame. + """ + await super().push_frame(frame, direction) + if isinstance(frame, (TTSStoppedFrame, StartInterruptionFrame)): + self._started = False 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( @@ -300,13 +295,12 @@ class AsyncAITTSService(AudioContextWordTTSService): sample_rate=self.sample_rate, num_channels=1, ) - await self.append_to_audio_context(context_id, frame) + await self.push_frame(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}") @@ -317,7 +311,7 @@ class AsyncAITTSService(AudioContextWordTTSService): self.reset_watchdog() await asyncio.sleep(KEEPALIVE_SLEEP) try: - if self._websocket and self._websocket.open: + if self._websocket and self._websocket.state is State.OPEN: keepalive_message = {"transcript": " "} logger.trace("Sending keepalive message") await self._websocket.send(json.dumps(keepalive_message)) @@ -338,15 +332,14 @@ class AsyncAITTSService(AudioContextWordTTSService): logger.debug(f"{self}: Generating TTS [{text}]") try: - if not self._websocket or self._websocket.closed: + if not self._websocket or self._websocket.state is State.CLOSED: await self._connect() - if not self._context_id: + if not self._started: await self.start_ttfb_metrics() yield TTSStartedFrame() - self._context_id = self._global_context_id - await self.create_audio_context(self._context_id) - + self._started = True + msg = self._build_msg(text=text) try: @@ -387,7 +380,7 @@ class AsyncAIHttpTTSService(TTSService): model: str = "asyncflow_v2.0", url: str = "https://api.async.ai", version: str = "v1", - sample_rate: int = 32000, + sample_rate: Optional[int] = None, encoding: str = "pcm_s16le", container: str = "raw", params: Optional[InputParams] = None, @@ -418,7 +411,7 @@ class AsyncAIHttpTTSService(TTSService): "output_format": { "container": container, "encoding": encoding, - "sample_rate": sample_rate, + "sample_rate": 0, }, "language": self.language_to_service_language(params.language) if params.language @@ -455,6 +448,7 @@ class AsyncAIHttpTTSService(TTSService): frame: The start frame containing initialization parameters. """ await super().start(frame) + self._settings["output_format"]["sample_rate"] = self.sample_rate @traced_tts async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: From 39b00f52695b8051fa65d955fbe39e25cd6eae9a Mon Sep 17 00:00:00 2001 From: Ashot Date: Tue, 29 Jul 2025 18:20:50 +0400 Subject: [PATCH 4/4] chore: address review comments --- CHANGELOG.md | 6 ++-- .../07ac-interruptible-asyncai-http.py | 6 ++-- .../07ac-interruptible-asyncai.py | 7 ++-- src/pipecat/services/asyncai/__init__.py | 7 ---- src/pipecat/services/asyncai/tts.py | 34 +++++++++---------- 5 files changed, 26 insertions(+), 34 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 18c433fe3..66c145670 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,9 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added -- **async.ai TTS integration** (https://async.ai/) - - `AsyncAITTSService` – streaming / interruptible TTS over WebSocket. - - `AsyncAIHttpTTSService` – streaming TTS over HTTP. +- Added Async.ai TTS integration (https://async.ai/) + - `AsyncAITTSService` – WebSocket-based streaming TTS with interruption support + - `AsyncAIHttpTTSService` – HTTP-based streaming TTS service - Example scripts: - `examples/foundational/07ac-interruptible-asyncai.py` (WebSocket demo) - `examples/foundational/07ac-interruptible-asyncai-http.py` (HTTP demo) diff --git a/examples/foundational/07ac-interruptible-asyncai-http.py b/examples/foundational/07ac-interruptible-asyncai-http.py index 415976d61..63710db14 100644 --- a/examples/foundational/07ac-interruptible-asyncai-http.py +++ b/examples/foundational/07ac-interruptible-asyncai-http.py @@ -16,8 +16,8 @@ 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.deepgram.stt import DeepgramSTTService from pipecat.services.openai.llm import OpenAILLMService from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams @@ -53,11 +53,11 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # Create an HTTP session async with aiohttp.ClientSession() as session: - stt = OpenAISTTService(api_key=os.getenv("OPENAI_API_KEY")) + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) tts = AsyncAIHttpTTSService( api_key=os.getenv("ASYNCAI_API_KEY", ""), - voice_id=os.getenv("ASYNCAI_VOICE_ID", ""), + voice_id=os.getenv("ASYNCAI_VOICE_ID", "e0f39dc4-f691-4e78-bba5-5c636692cc04"), aiohttp_session=session, ) diff --git a/examples/foundational/07ac-interruptible-asyncai.py b/examples/foundational/07ac-interruptible-asyncai.py index 49479999a..a27a8ed2b 100644 --- a/examples/foundational/07ac-interruptible-asyncai.py +++ b/examples/foundational/07ac-interruptible-asyncai.py @@ -9,13 +9,14 @@ 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.deepgram.stt import DeepgramSTTService from pipecat.services.openai.llm import OpenAILLMService from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams @@ -49,11 +50,11 @@ transport_params = { 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")) + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) tts = AsyncAITTSService( api_key=os.getenv("ASYNCAI_API_KEY", ""), - voice_id=os.getenv("ASYNCAI_VOICE_ID", ""), + voice_id=os.getenv("ASYNCAI_VOICE_ID", "e0f39dc4-f691-4e78-bba5-5c636692cc04"), ) llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) diff --git a/src/pipecat/services/asyncai/__init__.py b/src/pipecat/services/asyncai/__init__.py index 9542e4cd6..e69de29bb 100644 --- a/src/pipecat/services/asyncai/__init__.py +++ b/src/pipecat/services/asyncai/__init__.py @@ -1,7 +0,0 @@ -import sys - -from pipecat.services import DeprecatedModuleProxy - -from .tts import * - -sys.modules[__name__] = DeprecatedModuleProxy(globals(), "asyncai", "asyncai.tts") diff --git a/src/pipecat/services/asyncai/tts.py b/src/pipecat/services/asyncai/tts.py index d7d3dce00..e37aa20fa 100644 --- a/src/pipecat/services/asyncai/tts.py +++ b/src/pipecat/services/asyncai/tts.py @@ -6,15 +6,14 @@ """Async text-to-speech service implementations.""" +import asyncio import base64 import json -import uuid from typing import AsyncGenerator, Optional +import aiohttp from loguru import logger from pydantic import BaseModel -import asyncio -import aiohttp from pipecat.frames.frames import ( CancelFrame, @@ -103,7 +102,8 @@ class AsyncAITTSService(InterruptibleTTSService): Args: api_key: Async API key. - voice_id: ID of the voice to use for synthesis. + voice_id: UUID of the voice to use for synthesis. See docs for a full list: + https://docs.async.ai/list-voices-16699698e0 version: Async API version. url: WebSocket URL for Async TTS API. model: TTS model to use (e.g., "asyncflow_v2.0"). @@ -138,7 +138,7 @@ class AsyncAITTSService(InterruptibleTTSService): if params.language else "en", } - + self.set_model_name(model) self.set_voice(voice_id) @@ -165,15 +165,10 @@ class AsyncAITTSService(InterruptibleTTSService): """ return language_to_async_language(language) - def _build_msg( - self, text: str = "", force: bool = False - ) -> str: - msg = { - "transcript": text, - "force": force - } + def _build_msg(self, text: str = "", force: bool = False) -> str: + msg = {"transcript": text, "force": force} return json.dumps(msg) - + async def start(self, frame: StartFrame): """Start the Async TTS service. @@ -203,7 +198,7 @@ class AsyncAITTSService(InterruptibleTTSService): await self._disconnect() async def _connect(self): - await self._connect_websocket() + 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)) @@ -215,7 +210,7 @@ class AsyncAITTSService(InterruptibleTTSService): 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 @@ -268,7 +263,7 @@ class AsyncAITTSService(InterruptibleTTSService): logger.trace(f"{self}: flushing audio") msg = self._build_msg(text=" ", force=True) await self._websocket.send(msg) - + async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM): """Push a frame downstream with special handling for stop conditions. @@ -339,7 +334,7 @@ class AsyncAITTSService(InterruptibleTTSService): await self.start_ttfb_metrics() yield TTSStartedFrame() self._started = True - + msg = self._build_msg(text=text) try: @@ -355,6 +350,7 @@ class AsyncAITTSService(InterruptibleTTSService): except Exception as e: logger.error(f"{self} exception: {e}") + class AsyncAIHttpTTSService(TTSService): """HTTP-based Async TTS service. @@ -369,8 +365,9 @@ class AsyncAIHttpTTSService(TTSService): Parameters: language: Language to use for synthesis. """ + language: Optional[Language] = Language.EN - + def __init__( self, *, @@ -391,6 +388,7 @@ class AsyncAIHttpTTSService(TTSService): Args: api_key: Async API key. voice_id: ID of the voice to use for synthesis. + aiohttp_session: An aiohttp session for making HTTP requests. model: TTS model to use (e.g., "asyncflow_v2.0"). url: Base URL for Async API. version: API version string for Async API.