diff --git a/CHANGELOG.md b/CHANGELOG.md index ccf5dd5eb..53a2871aa 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,15 +9,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added -- Add support for more languages to ElevenLabs (Arabic, Croatian, Filipino, +- Pipecat now supports Python 3.13. We had a dependency on the `audioop` package + which was deprecated and now removed on Python 3.13. We are now using + `audioop-lts` (https://github.com/AbstractUmbra/audioop) to provide the same + functionality. + +- Added support for more languages to ElevenLabs (Arabic, Croatian, Filipino, Tamil) and PlayHT (Afrikans, Albanian, Amharic, Arabic, Bengali, Croatian, Galician, Hebrew, Mandarin, Serbian, Tagalog, Urdu, Xhosa). ### Changed -- Changed: Room expiration (`exp`) in `DailyRoomProperties` is now optional - (None) by default instead of automatically setting a 5-minute expiration - time. You must explicitly set expiration time if desired. +- Room expiration (`exp`) in `DailyRoomProperties` is now optional (`None`) by + default instead of automatically setting a 5-minute expiration time. You must + explicitly set expiration time if desired. ### Deprecated @@ -33,6 +38,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 only be pushed downstream after the audio generated from `TTSSpeakFrame` has been spoken. +## [0.0.51] - 2024-12-16 + +### Fixed + +- Fixed an issue in websocket-based TTS services that was causing infinite + reconnections (Cartesia, ElevenLabs, PlayHT and LMNT). + ## [0.0.50] - 2024-12-11 ### Added diff --git a/README.md b/README.md index 86e3ddd3e..f63d85167 100644 --- a/README.md +++ b/README.md @@ -55,19 +55,19 @@ pip install "pipecat-ai[option,...]" Available options include: -| Category | Services | Install Command Example | -| ------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------- | -| Speech-to-Text | [AssemblyAI](https://docs.pipecat.ai/api-reference/services/stt/assemblyai), [Azure](https://docs.pipecat.ai/api-reference/services/stt/azure), [Deepgram](https://docs.pipecat.ai/api-reference/services/stt/deepgram), [Gladia](https://docs.pipecat.ai/api-reference/services/stt/gladia), [Whisper](https://docs.pipecat.ai/api-reference/services/stt/whisper) | `pip install "pipecat-ai[deepgram]"` | -| LLMs | [Anthropic](https://docs.pipecat.ai/api-reference/services/llm/anthropic), [Azure](https://docs.pipecat.ai/api-reference/services/llm/azure), [Fireworks AI](https://docs.pipecat.ai/api-reference/services/llm/fireworks), [Gemini](https://docs.pipecat.ai/api-reference/services/llm/gemini), [Grok](https://docs.pipecat.ai/api-reference/services/llm/grok), [Groq](https://docs.pipecat.ai/api-reference/services/llm/groq), [NVIDIA NIM](https://docs.pipecat.ai/api-reference/services/llm/nim), [Ollama](https://docs.pipecat.ai/api-reference/services/llm/ollama), [OpenAI](https://docs.pipecat.ai/api-reference/services/llm/openai), [Together AI](https://docs.pipecat.ai/api-reference/services/llm/together) | `pip install "pipecat-ai[openai]"` | -| Text-to-Speech | [AWS](https://docs.pipecat.ai/api-reference/services/tts/aws), [Azure](https://docs.pipecat.ai/api-reference/services/tts/azure), [Cartesia](https://docs.pipecat.ai/api-reference/services/tts/cartesia), [Deepgram](https://docs.pipecat.ai/api-reference/services/tts/deepgram), [ElevenLabs](https://docs.pipecat.ai/api-reference/services/tts/elevenlabs), [Google](https://docs.pipecat.ai/api-reference/services/tts/google), [LMNT](https://docs.pipecat.ai/api-reference/services/tts/lmnt), [OpenAI](https://docs.pipecat.ai/api-reference/services/tts/openai), [PlayHT](https://docs.pipecat.ai/api-reference/services/tts/playht), [Rime](https://docs.pipecat.ai/api-reference/services/tts/rime), [XTTS](https://docs.pipecat.ai/api-reference/services/tts/xtts) | `pip install "pipecat-ai[cartesia]"` | -| Speech-to-Speech | [Gemini Multimodal Live](https://docs.pipecat.ai/server/services/s2s/gemini), [OpenAI Realtime](https://docs.pipecat.ai/api-reference/services/s2s/openai) | `pip install "pipecat-ai[openai]"` | -| Transport | [Daily (WebRTC)](https://docs.pipecat.ai/api-reference/services/transport/daily), WebSocket, Local | `pip install "pipecat-ai[daily]"` | -| Video | [Tavus](https://docs.pipecat.ai/api-reference/services/video/tavus), [Simli](https://docs.pipecat.ai/api-reference/services/video/simli) | `pip install "pipecat-ai[tavus,simli]"` | -| Vision & Image | [Moondream](https://docs.pipecat.ai/api-reference/services/vision/moondream), [fal](https://docs.pipecat.ai/api-reference/services/image-generation/fal) | `pip install "pipecat-ai[moondream]"` | -| Audio Processing | [Silero VAD](https://docs.pipecat.ai/api-reference/utilities/audio/silero-vad-analyzer), [Krisp](https://docs.pipecat.ai/api-reference/utilities/audio/krisp-filter), [Noisereduce](https://docs.pipecat.ai/api-reference/utilities/audio/noisereduce-filter) | `pip install "pipecat-ai[silero]"` | -| Analytics & Metrics | [Canonical AI](https://docs.pipecat.ai/api-reference/services/analytics/canonical), [Sentry](https://docs.pipecat.ai/api-reference/services/analytics/sentry) | `pip install "pipecat-ai[canonical]"` | +| Category | Services | Install Command Example | +| ------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------- | +| Speech-to-Text | [AssemblyAI](https://docs.pipecat.ai/server/services/stt/assemblyai), [Azure](https://docs.pipecat.ai/server/services/stt/azure), [Deepgram](https://docs.pipecat.ai/server/services/stt/deepgram), [Gladia](https://docs.pipecat.ai/server/services/stt/gladia), [Whisper](https://docs.pipecat.ai/server/services/stt/whisper) | `pip install "pipecat-ai[deepgram]"` | +| LLMs | [Anthropic](https://docs.pipecat.ai/server/services/llm/anthropic), [Azure](https://docs.pipecat.ai/server/services/llm/azure), [Fireworks AI](https://docs.pipecat.ai/server/services/llm/fireworks), [Gemini](https://docs.pipecat.ai/server/services/llm/gemini), [Grok](https://docs.pipecat.ai/server/services/llm/grok), [Groq](https://docs.pipecat.ai/server/services/llm/groq), [NVIDIA NIM](https://docs.pipecat.ai/server/services/llm/nim), [Ollama](https://docs.pipecat.ai/server/services/llm/ollama), [OpenAI](https://docs.pipecat.ai/server/services/llm/openai), [Together AI](https://docs.pipecat.ai/server/services/llm/together) | `pip install "pipecat-ai[openai]"` | +| Text-to-Speech | [AWS](https://docs.pipecat.ai/server/services/tts/aws), [Azure](https://docs.pipecat.ai/server/services/tts/azure), [Cartesia](https://docs.pipecat.ai/server/services/tts/cartesia), [Deepgram](https://docs.pipecat.ai/server/services/tts/deepgram), [ElevenLabs](https://docs.pipecat.ai/server/services/tts/elevenlabs), [Google](https://docs.pipecat.ai/server/services/tts/google), [LMNT](https://docs.pipecat.ai/server/services/tts/lmnt), [OpenAI](https://docs.pipecat.ai/server/services/tts/openai), [PlayHT](https://docs.pipecat.ai/server/services/tts/playht), [Rime](https://docs.pipecat.ai/server/services/tts/rime), [XTTS](https://docs.pipecat.ai/server/services/tts/xtts) | `pip install "pipecat-ai[cartesia]"` | +| Speech-to-Speech | [Gemini Multimodal Live](https://docs.pipecat.ai/server/services/s2s/gemini), [OpenAI Realtime](https://docs.pipecat.ai/server/services/s2s/openai) | `pip install "pipecat-ai[openai]"` | +| Transport | [Daily (WebRTC)](https://docs.pipecat.ai/server/services/transport/daily), WebSocket, Local | `pip install "pipecat-ai[daily]"` | +| Video | [Tavus](https://docs.pipecat.ai/server/services/video/tavus), [Simli](https://docs.pipecat.ai/server/services/video/simli) | `pip install "pipecat-ai[tavus,simli]"` | +| Vision & Image | [Moondream](https://docs.pipecat.ai/server/services/vision/moondream), [fal](https://docs.pipecat.ai/server/services/image-generation/fal) | `pip install "pipecat-ai[moondream]"` | +| Audio Processing | [Silero VAD](https://docs.pipecat.ai/server/utilities/audio/silero-vad-analyzer), [Krisp](https://docs.pipecat.ai/server/utilities/audio/krisp-filter), [Noisereduce](https://docs.pipecat.ai/server/utilities/audio/noisereduce-filter) | `pip install "pipecat-ai[silero]"` | +| Analytics & Metrics | [Canonical AI](https://docs.pipecat.ai/server/services/analytics/canonical), [Sentry](https://docs.pipecat.ai/server/services/analytics/sentry) | `pip install "pipecat-ai[canonical]"` | -📚 [View full services documentation →](https://docs.pipecat.ai/api-reference/services/supported-services) +📚 [View full services documentation →](https://docs.pipecat.ai/server/services/supported-services) ## Code examples diff --git a/dev-requirements.txt b/dev-requirements.txt index 92b6ec4d3..0ed6d9b05 100644 --- a/dev-requirements.txt +++ b/dev-requirements.txt @@ -1,8 +1,9 @@ -build~=1.2.1 -grpcio-tools~=1.65.4 +build~=1.2.2 +grpcio-tools~=1.68.1 pip-tools~=7.4.1 -pyright~=1.1.376 -pytest~=8.3.2 -ruff~=0.6.7 -setuptools~=72.2.0 +pyright~=1.1.390 +pytest~=8.3.4 +ruff~=0.8.3 +setuptools~=75.6.0 setuptools_scm~=8.1.0 +python-dotenv~=1.0.1 diff --git a/examples/foundational/14j-function-calling-nim.py b/examples/foundational/14j-function-calling-nim.py index 68bb3b52e..4059bc386 100644 --- a/examples/foundational/14j-function-calling-nim.py +++ b/examples/foundational/14j-function-calling-nim.py @@ -65,7 +65,7 @@ async def main(): ) llm = NimLLMService( - api_key=os.getenv("NVIDIA_API_KEY"), model="meta/llama-3.1-405b-instruct" + api_key=os.getenv("NVIDIA_API_KEY"), model="meta/llama-3.3-70b-instruct" ) # Register a function_name of None to get all functions # sent to the same callback with an additional function_name parameter. @@ -76,18 +76,18 @@ async def main(): type="function", function={ "name": "get_current_weather", - "description": "Get the current weather", + "description": "Returns the current weather at a location, if one is specified, and defaults to the user's location.", "parameters": { "type": "object", "properties": { "location": { "type": "string", - "description": "The city and state, e.g. San Francisco, CA", + "description": "The location to find the weather of, or if not provided, it's the default location.", }, "format": { "type": "string", "enum": ["celsius", "fahrenheit"], - "description": "The temperature unit to use. Infer this from the users location.", + "description": "Whether to use SI or USCS units (celsius or fahrenheit).", }, }, "required": ["location", "format"], diff --git a/examples/simple-chatbot/README.md b/examples/simple-chatbot/README.md index e9b06855d..4e23c9fae 100644 --- a/examples/simple-chatbot/README.md +++ b/examples/simple-chatbot/README.md @@ -24,12 +24,12 @@ This repository demonstrates a simple AI chatbot with real-time audio/video inte 2. **JavaScript** - - Basic implementation using [Pipecat JavaScript SDK](https://docs.pipecat.ai/client/reference/js/introduction) + - Basic implementation using [Pipecat JavaScript SDK](https://docs.pipecat.ai/client/js/introduction) - No framework dependencies - Good for learning the fundamentals 3. **React** - - Basic impelmentation using [Pipecat React SDK](https://docs.pipecat.ai/client/reference/react/introduction) + - Basic impelmentation using [Pipecat React SDK](https://docs.pipecat.ai/client/react/introduction) - Demonstrates the basic client principles with Pipecat React ## Quick Start diff --git a/examples/simple-chatbot/examples/javascript/README.md b/examples/simple-chatbot/examples/javascript/README.md index 74525c1c1..fdd97ce27 100644 --- a/examples/simple-chatbot/examples/javascript/README.md +++ b/examples/simple-chatbot/examples/javascript/README.md @@ -1,6 +1,6 @@ # JavaScript Implementation -Basic implementation using the [Pipecat JavaScript SDK](https://docs.pipecat.ai/client/reference/js/introduction). +Basic implementation using the [Pipecat JavaScript SDK](https://docs.pipecat.ai/client/js/introduction). ## Setup diff --git a/examples/simple-chatbot/examples/react/README.md b/examples/simple-chatbot/examples/react/README.md index 44775a083..892763d18 100644 --- a/examples/simple-chatbot/examples/react/README.md +++ b/examples/simple-chatbot/examples/react/README.md @@ -1,6 +1,6 @@ # React Implementation -Basic implementation using the [Pipecat React SDK](https://docs.pipecat.ai/client/reference/react/introduction). +Basic implementation using the [Pipecat React SDK](https://docs.pipecat.ai/client/react/introduction). ## Setup diff --git a/examples/studypal/env.example b/examples/studypal/env.example index 69245a3d6..6acef284b 100644 --- a/examples/studypal/env.example +++ b/examples/studypal/env.example @@ -1,4 +1,4 @@ -DAILY_SAMPLE_ROOM_URL= # Follow instructions here and put your https://YOURDOMAIN.daily.co/YOURROOM (Instructions: https://docs.pipecat.ai/quickstart#preparing-your-environment) +DAILY_SAMPLE_ROOM_URL= # Follow instructions here and put your https://YOURDOMAIN.daily.co/YOURROOM (Instructions: https://docs.pipecat.ai/getting-started/installation) DAILY_API_KEY= # Create here: https://dashboard.daily.co/developers OPENAI_API_KEY= # Create here: https://platform.openai.com/docs/overview CARTESIA_API_KEY= # Create here: https://play.cartesia.ai/console diff --git a/pyproject.toml b/pyproject.toml index 862981c7b..1a43b1b89 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -21,14 +21,17 @@ classifiers = [ ] dependencies = [ "aiohttp~=3.11.10", - "loguru~=0.7.2", + "audioop-lts~=0.2.1; python_version>='3.13'", + "loguru~=0.7.3", "Markdown~=3.7", - "numpy~=1.26.4", - "Pillow~=10.4.0", + "numpy~=2.1.3", + "numba~=0.61.0rc1", + "Pillow~=11.0.0", "protobuf~=5.29.1", - "pydantic~=2.8.2", + "pydantic~=2.10.3", "pyloudnorm~=0.1.1", "resampy~=0.4.3", + "tenacity~=9.0.0" ] [project.urls] @@ -45,7 +48,6 @@ cartesia = [ "cartesia~=1.0.13", "websockets~=13.1" ] daily = [ "daily-python~=0.13.0" ] deepgram = [ "deepgram-sdk~=3.7.7" ] elevenlabs = [ "websockets~=13.1" ] -examples = [ "python-dotenv~=1.0.1", "flask~=3.0.3", "flask_cors~=4.0.1" ] fal = [ "fal-client~=0.4.1" ] gladia = [ "websockets~=13.1" ] google = [ "google-generativeai~=0.8.3", "google-cloud-texttospeech~=2.21.1" ] @@ -55,15 +57,15 @@ gstreamer = [ "pygobject~=3.48.2" ] fireworks = [ "openai~=1.57.2" ] krisp = [ "pipecat-ai-krisp~=0.3.0" ] langchain = [ "langchain~=0.2.14", "langchain-community~=0.2.12", "langchain-openai~=0.1.20" ] -livekit = [ "livekit~=0.18.2", "livekit-api~=0.8.0", "tenacity~=8.5.0" ] +livekit = [ "livekit~=0.17.5", "livekit-api~=0.7.1" ] lmnt = [ "lmnt~=1.1.4" ] local = [ "pyaudio~=0.2.14" ] moondream = [ "einops~=0.8.0", "timm~=1.0.8", "transformers~=4.44.0" ] nim = [ "openai~=1.57.2" ] noisereduce = [ "noisereduce~=3.0.3" ] openai = [ "openai~=1.57.2", "websockets~=13.1", "python-deepcompare~=1.0.1" ] -openpipe = [ "openpipe~=4.38.0" ] -playht = [ "pyht~=0.1.8", "websockets~=13.1" ] +openpipe = [ "openpipe~=4.40.0" ] +playht = [ "pyht~=0.1.9", "websockets~=13.1" ] riva = [ "nvidia-riva-client~=2.17.0" ] silero = [ "onnxruntime~=1.20.1" ] simli = [ "simli-ai~=0.1.7"] diff --git a/src/pipecat/processors/metrics/frame_processor_metrics.py b/src/pipecat/processors/metrics/frame_processor_metrics.py index a22639239..2c0099989 100644 --- a/src/pipecat/processors/metrics/frame_processor_metrics.py +++ b/src/pipecat/processors/metrics/frame_processor_metrics.py @@ -1,3 +1,9 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + import time from pipecat.frames.frames import MetricsFrame diff --git a/src/pipecat/processors/metrics/sentry.py b/src/pipecat/processors/metrics/sentry.py index e37dd9d44..6cc6d1103 100644 --- a/src/pipecat/processors/metrics/sentry.py +++ b/src/pipecat/processors/metrics/sentry.py @@ -1,3 +1,9 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + import time from loguru import logger @@ -29,8 +35,9 @@ class SentryMetrics(FrameProcessorMetrics): description=f"TTFB for {self._processor_name()}", start_timestamp=self._start_ttfb_time, ) - logger.debug(f"Sentry Span ID: {self._ttfb_metrics_span.span_id} Description: { - self._ttfb_metrics_span.description} started.") + logger.debug( + f"Sentry Span ID: {self._ttfb_metrics_span.span_id} Description: {self._ttfb_metrics_span.description} started." + ) self._should_report_ttfb = not report_only_initial_ttfb async def stop_ttfb_metrics(self): @@ -46,8 +53,9 @@ class SentryMetrics(FrameProcessorMetrics): description=f"Processing for {self._processor_name()}", start_timestamp=self._start_processing_time, ) - logger.debug(f"Sentry Span ID: {self._processing_metrics_span.span_id} Description: { - self._processing_metrics_span.description} started.") + logger.debug( + f"Sentry Span ID: {self._processing_metrics_span.span_id} Description: {self._processing_metrics_span.description} started." + ) async def stop_processing_metrics(self): stop_time = time.time() diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index 8683fd29a..88fbf29d4 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -12,6 +12,8 @@ from typing import AsyncGenerator, List, Optional, Union from loguru import logger from pydantic import BaseModel +from tenacity import AsyncRetrying, RetryCallState, stop_after_attempt, wait_exponential + from pipecat.frames.frames import ( BotStoppedSpeakingFrame, @@ -239,52 +241,64 @@ class CartesiaTTSService(WordTTSService): msg = self._build_msg(text="", continue_transcript=False) await self._websocket.send(msg) + async def _receive_messages(self): + async for message in self._get_websocket(): + msg = json.loads(message) + if not msg or msg["context_id"] != self._context_id: + continue + if msg["type"] == "done": + await self.stop_ttfb_metrics() + # Unset _context_id but not the _context_id_start_timestamp + # because we are likely still playing out audio and need the + # timestamp to set send context frames. + self._context_id = None + await self.add_word_timestamps( + [("TTSStoppedFrame", 0), ("LLMFullResponseEndFrame", 0), ("Reset", 0)] + ) + elif msg["type"] == "timestamps": + await self.add_word_timestamps( + list(zip(msg["word_timestamps"]["words"], msg["word_timestamps"]["start"])) + ) + elif msg["type"] == "chunk": + await self.stop_ttfb_metrics() + self.start_word_timestamps() + frame = TTSAudioRawFrame( + audio=base64.b64decode(msg["data"]), + sample_rate=self._settings["output_format"]["sample_rate"], + num_channels=1, + ) + await self.push_frame(frame) + elif msg["type"] == "error": + 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["error"]}')) + else: + logger.error(f"{self} error, unknown message type: {msg}") + + async def _reconnect_websocket(self, retry_state: RetryCallState): + logger.warning(f"{self} 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 message in self._get_websocket(): - msg = json.loads(message) - if not msg or msg["context_id"] != self._context_id: - continue - if msg["type"] == "done": - await self.stop_ttfb_metrics() - # Unset _context_id but not the _context_id_start_timestamp - # because we are likely still playing out audio and need the - # timestamp to set send context frames. - self._context_id = None - await self.add_word_timestamps( - [("TTSStoppedFrame", 0), ("LLMFullResponseEndFrame", 0), ("Reset", 0)] - ) - elif msg["type"] == "timestamps": - await self.add_word_timestamps( - list( - zip( - msg["word_timestamps"]["words"], msg["word_timestamps"]["start"] - ) - ) - ) - elif msg["type"] == "chunk": - await self.stop_ttfb_metrics() - self.start_word_timestamps() - frame = TTSAudioRawFrame( - audio=base64.b64decode(msg["data"]), - sample_rate=self._settings["output_format"]["sample_rate"], - num_channels=1, - ) - await self.push_frame(frame) - elif msg["type"] == "error": - 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["error"]}')) - else: - logger.error(f"{self} error, unknown message type: {msg}") + 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: - logger.error(f"{self} exception: {e}") - await self._disconnect_websocket() - await self._connect_websocket() + message = f"{self} 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) diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index b829f4945..011c64308 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -11,11 +11,13 @@ from typing import Any, AsyncGenerator, Dict, List, Literal, Mapping, Optional, from loguru import logger from pydantic import BaseModel, model_validator +from tenacity import AsyncRetrying, RetryCallState, stop_after_attempt, wait_exponential from pipecat.frames.frames import ( BotStoppedSpeakingFrame, CancelFrame, EndFrame, + ErrorFrame, Frame, LLMFullResponseEndFrame, StartFrame, @@ -352,28 +354,44 @@ class ElevenLabsTTSService(WordTTSService): except Exception as e: logger.error(f"{self} error closing websocket: {e}") + async def _receive_messages(self): + async for message in self._websocket: + msg = json.loads(message) + if msg.get("audio"): + await self.stop_ttfb_metrics() + self.start_word_timestamps() + + audio = base64.b64decode(msg["audio"]) + frame = TTSAudioRawFrame(audio, self._settings["sample_rate"], 1) + await self.push_frame(frame) + if msg.get("alignment"): + word_times = calculate_word_times(msg["alignment"], self._cumulative_time) + await self.add_word_timestamps(word_times) + self._cumulative_time = word_times[-1][1] + + async def _reconnect_websocket(self, retry_state: RetryCallState): + logger.warning(f"{self} 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 message in self._websocket: - msg = json.loads(message) - if msg.get("audio"): - await self.stop_ttfb_metrics() - self.start_word_timestamps() - - audio = base64.b64decode(msg["audio"]) - frame = TTSAudioRawFrame(audio, self._settings["sample_rate"], 1) - await self.push_frame(frame) - if msg.get("alignment"): - word_times = calculate_word_times(msg["alignment"], self._cumulative_time) - await self.add_word_timestamps(word_times) - self._cumulative_time = word_times[-1][1] + 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: - logger.error(f"{self} exception: {e}") - await self._disconnect_websocket() - await self._connect_websocket() + message = f"{self} error receiving messages: {e}" + logger.error(message) + await self.push_error(ErrorFrame(message, fatal=True)) + break async def _keepalive_task_handler(self): while True: diff --git a/src/pipecat/services/grok.py b/src/pipecat/services/grok.py index 505dfcca5..a6a1b3a64 100644 --- a/src/pipecat/services/grok.py +++ b/src/pipecat/services/grok.py @@ -5,11 +5,102 @@ # +import json +from dataclasses import dataclass + from loguru import logger from pipecat.metrics.metrics import LLMTokenUsage -from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext -from pipecat.services.openai import OpenAILLMService +from pipecat.processors.aggregators.openai_llm_context import ( + OpenAILLMContext, + OpenAILLMContextFrame, +) +from pipecat.services.openai import ( + OpenAIAssistantContextAggregator, + OpenAILLMService, + OpenAIUserContextAggregator, +) + + +class GrokAssistantContextAggregator(OpenAIAssistantContextAggregator): + """Custom assistant context aggregator for Grok that handles empty content requirement.""" + + async def _push_aggregation(self): + if not ( + self._aggregation or self._function_call_result or self._pending_image_frame_message + ): + return + + run_llm = False + + aggregation = self._aggregation + self._reset() + + try: + if self._function_call_result: + frame = self._function_call_result + self._function_call_result = None + if frame.result: + # Grok requires an empty content field for function calls + self._context.add_message( + { + "role": "assistant", + "content": "", # Required by Grok + "tool_calls": [ + { + "id": frame.tool_call_id, + "function": { + "name": frame.function_name, + "arguments": json.dumps(frame.arguments), + }, + "type": "function", + } + ], + } + ) + self._context.add_message( + { + "role": "tool", + "content": json.dumps(frame.result), + "tool_call_id": frame.tool_call_id, + } + ) + # Only run the LLM if there are no more function calls in progress. + run_llm = not bool(self._function_calls_in_progress) + else: + self._context.add_message({"role": "assistant", "content": aggregation}) + + if self._pending_image_frame_message: + frame = self._pending_image_frame_message + self._pending_image_frame_message = None + self._context.add_image_frame_message( + format=frame.user_image_raw_frame.format, + size=frame.user_image_raw_frame.size, + image=frame.user_image_raw_frame.image, + text=frame.text, + ) + run_llm = True + + if run_llm: + await self._user_context_aggregator.push_context_frame() + + frame = OpenAILLMContextFrame(self._context) + await self.push_frame(frame) + + except Exception as e: + logger.error(f"Error processing frame: {e}") + + +@dataclass +class GrokContextAggregatorPair: + _user: "OpenAIUserContextAggregator" + _assistant: "GrokAssistantContextAggregator" + + def user(self) -> "OpenAIUserContextAggregator": + return self._user + + def assistant(self) -> "GrokAssistantContextAggregator": + return self._assistant class GrokLLMService(OpenAILLMService): @@ -101,3 +192,13 @@ class GrokLLMService(OpenAILLMService): # Update completion tokens count if it has increased if tokens.completion_tokens > self._completion_tokens: self._completion_tokens = tokens.completion_tokens + + @staticmethod + def create_context_aggregator( + context: OpenAILLMContext, *, assistant_expect_stripped_words: bool = True + ) -> GrokContextAggregatorPair: + user = OpenAIUserContextAggregator(context) + assistant = GrokAssistantContextAggregator( + user, expect_stripped_words=assistant_expect_stripped_words + ) + return GrokContextAggregatorPair(_user=user, _assistant=assistant) diff --git a/src/pipecat/services/lmnt.py b/src/pipecat/services/lmnt.py index 04223a1a1..5393e6653 100644 --- a/src/pipecat/services/lmnt.py +++ b/src/pipecat/services/lmnt.py @@ -8,6 +8,7 @@ import asyncio from typing import AsyncGenerator from loguru import logger +from tenacity import AsyncRetrying, RetryCallState, stop_after_attempt, wait_exponential from pipecat.frames.frames import ( CancelFrame, @@ -159,31 +160,47 @@ class LmntTTSService(TTSService): except Exception as e: logger.error(f"{self} error closing connection: {e}") + async def _receive_messages(self): + async for msg in self._connection: + if "error" in msg: + logger.error(f'{self} error: {msg["error"]}') + await self.push_frame(TTSStoppedFrame()) + await self.stop_all_metrics() + await self.push_error(ErrorFrame(f'{self} error: {msg["error"]}')) + elif "audio" in msg: + await self.stop_ttfb_metrics() + frame = TTSAudioRawFrame( + audio=msg["audio"], + sample_rate=self._settings["output_format"]["sample_rate"], + num_channels=1, + ) + await self.push_frame(frame) + else: + logger.error(f"{self}: LMNT error, unknown message type: {msg}") + + async def _reconnect_websocket(self, retry_state: RetryCallState): + logger.warning(f"{self} reconnecting (attempt: {retry_state.attempt_number})") + await self._disconnect_lmnt() + await self._connect_lmnt() + async def _receive_task_handler(self): while True: try: - async for msg in self._connection: - if "error" in msg: - logger.error(f'{self} error: {msg["error"]}') - await self.push_frame(TTSStoppedFrame()) - await self.stop_all_metrics() - await self.push_error(ErrorFrame(f'{self} error: {msg["error"]}')) - elif "audio" in msg: - await self.stop_ttfb_metrics() - frame = TTSAudioRawFrame( - audio=msg["audio"], - sample_rate=self._settings["output_format"]["sample_rate"], - num_channels=1, - ) - await self.push_frame(frame) - else: - logger.error(f"{self}: LMNT error, unknown message type: {msg}") + 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: - logger.error(f"{self} exception: {e}") - await self._disconnect_lmnt() - await self._connect_lmnt() + message = f"{self} error receiving messages: {e}" + logger.error(message) + await self.push_error(ErrorFrame(message, fatal=True)) + break async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: logger.debug(f"Generating TTS: [{text}]") diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index 43ad16536..85e1a95f0 100644 --- a/src/pipecat/services/openai.py +++ b/src/pipecat/services/openai.py @@ -559,7 +559,6 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator): self._context.add_message( { "role": "assistant", - "content": "", # content field required for Grok function calling "tool_calls": [ { "id": frame.tool_call_id, diff --git a/src/pipecat/services/playht.py b/src/pipecat/services/playht.py index 00c5d4d34..43e1739c9 100644 --- a/src/pipecat/services/playht.py +++ b/src/pipecat/services/playht.py @@ -15,6 +15,7 @@ import aiohttp import websockets from loguru import logger from pydantic import BaseModel +from tenacity import AsyncRetrying, RetryCallState, stop_after_attempt, wait_exponential from pipecat.frames.frames import ( BotStoppedSpeakingFrame, @@ -234,35 +235,51 @@ class PlayHTTTSService(TTSService): await self.stop_all_metrics() self._request_id = None + async def _receive_messages(self): + async for message in self._get_websocket(): + if isinstance(message, bytes): + # Skip the WAV header message + if message.startswith(b"RIFF"): + continue + await self.stop_ttfb_metrics() + frame = TTSAudioRawFrame(message, self._settings["sample_rate"], 1) + await self.push_frame(frame) + else: + logger.debug(f"Received text message: {message}") + try: + msg = json.loads(message) + if "request_id" in msg and msg["request_id"] == self._request_id: + await self.push_frame(TTSStoppedFrame()) + self._request_id = None + elif "error" in msg: + logger.error(f"{self} error: {msg}") + await self.push_error(ErrorFrame(f'{self} error: {msg["error"]}')) + except json.JSONDecodeError: + logger.error(f"Invalid JSON message: {message}") + + async def _reconnect_websocket(self, retry_state: RetryCallState): + logger.warning(f"{self} 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 message in self._get_websocket(): - if isinstance(message, bytes): - # Skip the WAV header message - if message.startswith(b"RIFF"): - continue - await self.stop_ttfb_metrics() - frame = TTSAudioRawFrame(message, self._settings["sample_rate"], 1) - await self.push_frame(frame) - else: - logger.debug(f"Received text message: {message}") - try: - msg = json.loads(message) - if "request_id" in msg and msg["request_id"] == self._request_id: - await self.push_frame(TTSStoppedFrame()) - self._request_id = None - elif "error" in msg: - logger.error(f"{self} error: {msg}") - await self.push_error(ErrorFrame(f'{self} error: {msg["error"]}')) - except json.JSONDecodeError: - logger.error(f"Invalid JSON message: {message}") + 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: - logger.error(f"{self} exception in receive task: {e}") - await self._disconnect_websocket() - await self._connect_websocket() + message = f"{self} 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)