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.