From f2e9562f1b31f70f043b3617c3395616913c20c9 Mon Sep 17 00:00:00 2001 From: Ashot Date: Wed, 23 Jul 2025 14:44:21 +0400 Subject: [PATCH] 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()