chore: address review comments

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/)
+- Added Async.ai TTS integration (https://async.ai/)
-  - `AsyncAITTSService` – streaming / interruptible TTS over WebSocket.
+  - `AsyncAITTSService` – WebSocket-based streaming TTS with interruption support
-  - `AsyncAIHttpTTSService` – streaming TTS over HTTP.
+  - `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)

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,
         )
 

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"))

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")

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(
+    def _build_msg(self, text: str = "", force: bool = False) -> str:
-        self, text: str = "", force: bool = False
+        msg = {"transcript": text, "force": force}
-    ) -> 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.