diff --git a/CHANGELOG.md b/CHANGELOG.md index 0982d6b4c..58a64ae90 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 +- 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) + - Added `transcription_bucket` params support to the `DailyRESTHelper`. - Added a new TTS service, `InworldTTSService`. This service provides 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/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/07ac-interruptible-asyncai-http.py b/examples/foundational/07ac-interruptible-asyncai-http.py new file mode 100644 index 000000000..63710db14 --- /dev/null +++ b/examples/foundational/07ac-interruptible-asyncai-http.py @@ -0,0 +1,116 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +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.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 +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 = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + + tts = AsyncAIHttpTTSService( + api_key=os.getenv("ASYNCAI_API_KEY", ""), + voice_id=os.getenv("ASYNCAI_VOICE_ID", "e0f39dc4-f691-4e78-bba5-5c636692cc04"), + 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/07ac-interruptible-asyncai.py b/examples/foundational/07ac-interruptible-asyncai.py new file mode 100644 index 000000000..a27a8ed2b --- /dev/null +++ b/examples/foundational/07ac-interruptible-asyncai.py @@ -0,0 +1,112 @@ +# +# 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.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 +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 = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + + tts = AsyncAITTSService( + api_key=os.getenv("ASYNCAI_API_KEY", ""), + voice_id=os.getenv("ASYNCAI_VOICE_ID", "e0f39dc4-f691-4e78-bba5-5c636692cc04"), + ) + + 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/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 new file mode 100644 index 000000000..e69de29bb diff --git a/src/pipecat/services/asyncai/tts.py b/src/pipecat/services/asyncai/tts.py new file mode 100644 index 000000000..e37aa20fa --- /dev/null +++ b/src/pipecat/services/asyncai/tts.py @@ -0,0 +1,505 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Async text-to-speech service implementations.""" + +import asyncio +import base64 +import json +from typing import AsyncGenerator, Optional + +import aiohttp +from loguru import logger +from pydantic import BaseModel + +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 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]`.") + 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(InterruptibleTTSService): + """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: Optional[int] = None, + 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: 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"). + 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. + """ + super().__init__( + aggregate_sentences=aggregate_sentences, + push_text_frames=False, + pause_frame_processing=True, + push_stop_frames=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": 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._receive_task = None + self._keepalive_task = None + self._started = False + + 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) -> str: + 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) + self._settings["output_format"]["sample_rate"] = self.sample_rate + 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.state is State.OPEN: + return + logger.debug("Connecting to Async") + self._websocket = await websocket_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._websocket = None + self._started = False + + def _get_websocket(self): + if self._websocket: + return self._websocket + raise Exception("Websocket not connected") + + async def flush_audio(self): + """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) + + 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) + if not msg: + continue + + 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.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']}")) + 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.state is State.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.state is State.CLOSED: + await self._connect() + + if not self._started: + await self.start_ttfb_metrics() + yield TTSStartedFrame() + self._started = True + + 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 Async's 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: Optional[int] = None, + 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. + 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. + 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": 0, + }, + "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) + self._settings["output_format"]["sample_rate"] = self.sample_rate + + @traced_tts + async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + """Generate speech from text using Async's HTTP streaming API. + + Args: + text: The text to synthesize into speech. + + Yields: + Frame: Audio frames containing the synthesized speech. + """ + logger.debug(f"{self}: Generating TTS [{text}]") + + try: + voice_config = {"mode": "id", "id": self._voice_id} + await self.start_ttfb_metrics() + payload = { + "model_id": self._model_name, + "transcript": text, + "voice": voice_config, + "output_format": self._settings["output_format"], + "language": self._settings["language"], + } + yield TTSStartedFrame() + headers = { + "version": self._api_version, + "x-api-key": self._api_key, + "Content-Type": "application/json", + } + url = f"{self._base_url}/text_to_speech/streaming" + + async with self._session.post(url, json=payload, headers=headers) as response: + if response.status != 200: + error_text = await response.text() + 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()