diff --git a/pyproject.toml b/pyproject.toml index ee842a533..38bf6d902 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,8 +32,7 @@ dependencies = [ "protobuf~=5.29.3", "pydantic~=2.10.5", "pyloudnorm~=0.1.1", - "resampy~=0.4.3", - "tenacity~=9.0.0" + "resampy~=0.4.3" ] [project.urls] @@ -63,7 +62,7 @@ fireworks = [ "openai~=1.59.6" ] krisp = [ "pipecat-ai-krisp~=0.3.0" ] koala = [ "pvkoala~=2.0.3" ] langchain = [ "langchain~=0.3.14", "langchain-community~=0.3.14", "langchain-openai~=0.3.0" ] -livekit = [ "livekit~=0.19.1", "livekit-api~=0.8.1" ] +livekit = [ "livekit~=0.19.1", "livekit-api~=0.8.1", "tenacity~=9.0.0" ] lmnt = [ "websockets~=13.1" ] local = [ "pyaudio~=0.2.14" ] moondream = [ "einops~=0.8.0", "timm~=1.0.13", "transformers~=4.48.0" ] diff --git a/src/pipecat/services/fish.py b/src/pipecat/services/fish.py index 245497b1d..710ab04ec 100644 --- a/src/pipecat/services/fish.py +++ b/src/pipecat/services/fish.py @@ -4,13 +4,11 @@ # SPDX-License-Identifier: BSD 2-Clause License # -import asyncio import uuid from typing import AsyncGenerator, Literal, Optional from loguru import logger from pydantic import BaseModel -from tenacity import AsyncRetrying, RetryCallState, stop_after_attempt, wait_exponential from pipecat.frames.frames import ( BotStoppedSpeakingFrame, @@ -28,6 +26,7 @@ from pipecat.frames.frames import ( ) from pipecat.processors.frame_processor import FrameDirection from pipecat.services.ai_services import TTSService +from pipecat.services.websocket_service import WebsocketService from pipecat.transcriptions.language import Language try: @@ -44,7 +43,7 @@ except ModuleNotFoundError as e: FishAudioOutputFormat = Literal["opus", "mp3", "pcm", "wav"] -class FishAudioTTSService(TTSService): +class FishAudioTTSService(TTSService, WebsocketService): class InputParams(BaseModel): language: Optional[Language] = Language.EN latency: Optional[str] = "normal" # "normal" or "balanced" @@ -105,7 +104,9 @@ class FishAudioTTSService(TTSService): async def _connect(self): await self._connect_websocket() - self._receive_task = self.get_event_loop().create_task(self._receive_task_handler()) + self._receive_task = self.get_event_loop().create_task( + self._receive_task_handler(self.push_error) + ) async def _disconnect(self): await self._disconnect_websocket() @@ -169,30 +170,6 @@ class FishAudioTTSService(TTSService): except Exception as e: logger.error(f"Error processing message: {e}") - async def _reconnect_websocket(self, retry_state: RetryCallState): - logger.warning(f"Fish Audio reconnecting (attempt: {retry_state.attempt_number})") - await self._disconnect_websocket() - await self._connect_websocket() - - async def _receive_task_handler(self): - while True: - try: - async for attempt in AsyncRetrying( - stop=stop_after_attempt(3), - wait=wait_exponential(multiplier=1, min=4, max=10), - before_sleep=self._reconnect_websocket, - reraise=True, - ): - with attempt: - await self._receive_messages() - except asyncio.CancelledError: - break - except Exception as e: - message = f"Fish Audio error receiving messages: {e}" - logger.error(message) - await self.push_error(ErrorFrame(message, fatal=True)) - break - async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction)