From 5c0f5a1613b1b44bb0ff63231ea00436899bac5c Mon Sep 17 00:00:00 2001 From: Soof Golan Date: Mon, 2 Sep 2024 14:44:48 +0200 Subject: [PATCH 01/25] Generate ids with itertools.count Avoids the critical section with threading.Lock in favor of itertools.count. `count` objects are threadsafe, and their critical section is implemented in C and provide better performance that Python level locking. --- src/pipecat/utils/utils.py | 45 ++++++++++++++++++++++---------------- 1 file changed, 26 insertions(+), 19 deletions(-) diff --git a/src/pipecat/utils/utils.py b/src/pipecat/utils/utils.py index 0be73191f..e2df99389 100644 --- a/src/pipecat/utils/utils.py +++ b/src/pipecat/utils/utils.py @@ -3,32 +3,39 @@ # # SPDX-License-Identifier: BSD 2-Clause License # +import collections +import itertools -from threading import Lock - -_COUNTS = {} -_COUNTS_MUTEX = Lock() - -_ID = 0 -_ID_MUTEX = Lock() +_COUNTS = collections.defaultdict(itertools.count) +_ID = itertools.count() def obj_id() -> int: - global _ID, _ID_MUTEX - with _ID_MUTEX: - _ID += 1 - return _ID + """ + Generate a unique id for an object. + + >>> obj_id() + 0 + >>> obj_id() + 1 + >>> obj_id() + 2 + """ + return next(_ID) def obj_count(obj) -> int: - global _COUNTS, COUNTS_MUTEX - name = obj.__class__.__name__ - with _COUNTS_MUTEX: - if name not in _COUNTS: - _COUNTS[name] = 0 - else: - _COUNTS[name] += 1 - return _COUNTS[name] + """Generate a unique id for an object. + + >>> obj_count(object()) + 0 + >>> obj_count(object()) + 1 + >>> new_type = type('NewType', (object,), {}) + >>> obj_count(new_type()) + 0 + """ + return next(_COUNTS[obj.__class__.__name__]) def exp_smoothing(value: float, prev_value: float, factor: float) -> float: From 51cd7fd285e9cd0897d51b299a27ee77006b27cb Mon Sep 17 00:00:00 2001 From: Aashraya Date: Mon, 2 Sep 2024 23:36:38 +0530 Subject: [PATCH 02/25] twiliohandle interruption (#422) * add interuption handler in twilio serializer * fix autopep8 * revert ruff autoformatting * address pr comments * change interruption frame to user started frame in serializer * remove overrrident handle interrupt * remove unused import * change userstarted to interuption frame --- src/pipecat/serializers/twilio.py | 31 ++++++++++--------- .../transports/network/fastapi_websocket.py | 21 ++++++++++--- 2 files changed, 34 insertions(+), 18 deletions(-) diff --git a/src/pipecat/serializers/twilio.py b/src/pipecat/serializers/twilio.py index 8836fcd6d..583234ae4 100644 --- a/src/pipecat/serializers/twilio.py +++ b/src/pipecat/serializers/twilio.py @@ -9,7 +9,7 @@ import json from pydantic import BaseModel -from pipecat.frames.frames import AudioRawFrame, Frame +from pipecat.frames.frames import AudioRawFrame, Frame, StartInterruptionFrame from pipecat.serializers.base_serializer import FrameSerializer from pipecat.utils.audio import ulaw_to_pcm, pcm_to_ulaw @@ -28,22 +28,25 @@ class TwilioFrameSerializer(FrameSerializer): self._params = params def serialize(self, frame: Frame) -> str | bytes | None: - if not isinstance(frame, AudioRawFrame): - return None + if isinstance(frame, AudioRawFrame): + data = frame.audio - data = frame.audio - - serialized_data = pcm_to_ulaw(data, frame.sample_rate, self._params.twilio_sample_rate) - payload = base64.b64encode(serialized_data).decode("utf-8") - answer = { - "event": "media", - "streamSid": self._stream_sid, - "media": { - "payload": payload + serialized_data = pcm_to_ulaw( + data, frame.sample_rate, self._params.twilio_sample_rate) + payload = base64.b64encode(serialized_data).decode("utf-8") + answer = { + "event": "media", + "streamSid": self._stream_sid, + "media": { + "payload": payload + } } - } - return json.dumps(answer) + return json.dumps(answer) + + if isinstance(frame, StartInterruptionFrame): + answer = {"event": "clear", "streamSid": self._stream_sid} + return json.dumps(answer) def deserialize(self, data: str | bytes) -> Frame | None: message = json.loads(data) diff --git a/src/pipecat/transports/network/fastapi_websocket.py b/src/pipecat/transports/network/fastapi_websocket.py index 914870114..16c5e81fc 100644 --- a/src/pipecat/transports/network/fastapi_websocket.py +++ b/src/pipecat/transports/network/fastapi_websocket.py @@ -12,8 +12,8 @@ import wave from typing import Awaitable, Callable from pydantic.main import BaseModel -from pipecat.frames.frames import AudioRawFrame, CancelFrame, EndFrame, StartFrame -from pipecat.processors.frame_processor import FrameProcessor +from pipecat.frames.frames import AudioRawFrame, CancelFrame, EndFrame, Frame, StartFrame, StartInterruptionFrame +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.serializers.base_serializer import FrameSerializer from pipecat.transports.base_input import BaseInputTransport from pipecat.transports.base_output import BaseOutputTransport @@ -93,11 +93,18 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport): self._params = params self._websocket_audio_buffer = bytes() + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if isinstance(frame, StartInterruptionFrame): + await self._write_frame(frame) + async def write_raw_audio_frames(self, frames: bytes): self._websocket_audio_buffer += frames while len(self._websocket_audio_buffer) >= self._params.audio_frame_size: frame = AudioRawFrame( - audio=self._websocket_audio_buffer[:self._params.audio_frame_size], + audio=self._websocket_audio_buffer[: + self._params.audio_frame_size], sample_rate=self._params.audio_out_sample_rate, num_channels=self._params.audio_out_channels ) @@ -121,7 +128,13 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport): if payload and self._websocket.client_state == WebSocketState.CONNECTED: await self._websocket.send_text(payload) - self._websocket_audio_buffer = self._websocket_audio_buffer[self._params.audio_frame_size:] + self._websocket_audio_buffer = self._websocket_audio_buffer[ + self._params.audio_frame_size:] + + async def _write_frame(self, frame: Frame): + payload = self._params.serializer.serialize(frame) + if payload and self._websocket.client_state == WebSocketState.CONNECTED: + await self._websocket.send_text(payload) class FastAPIWebsocketTransport(BaseTransport): From e405d7af9fd2936fc1a42a55557a651cc0eab7ef Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 5 Sep 2024 11:12:52 -0700 Subject: [PATCH 03/25] services(elevenlabs): add elevenlabs package and use streaming --- examples/deployment/flyio-example/bot.py | 105 ++++++------- examples/dialin-chatbot/bot_daily.py | 116 +++++++------- examples/dialin-chatbot/bot_twilio.py | 148 +++++++++--------- .../foundational/05-sync-speech-and-image.py | 1 - .../05a-local-sync-speech-and-image.py | 1 - examples/foundational/06a-image-sync.py | 1 - .../07b-interruptible-langchain.py | 1 - examples/foundational/11-sound-effects.py | 1 - examples/simple-chatbot/bot.py | 1 - examples/storytelling-chatbot/src/bot.py | 1 - pyproject.toml | 1 + src/pipecat/services/cartesia.py | 2 +- src/pipecat/services/elevenlabs.py | 80 ++++++---- 13 files changed, 230 insertions(+), 229 deletions(-) diff --git a/examples/deployment/flyio-example/bot.py b/examples/deployment/flyio-example/bot.py index cc68f5522..c6380f6f3 100644 --- a/examples/deployment/flyio-example/bot.py +++ b/examples/deployment/flyio-example/bot.py @@ -1,5 +1,4 @@ import asyncio -import aiohttp import os import sys import argparse @@ -27,71 +26,69 @@ daily_api_url = os.getenv("DAILY_API_URL", "https://api.daily.co/v1") async def main(room_url: str, token: str): - async with aiohttp.ClientSession() as session: - transport = DailyTransport( - room_url, - token, - "Chatbot", - DailyParams( - api_url=daily_api_url, - api_key=daily_api_key, - audio_in_enabled=True, - audio_out_enabled=True, - camera_out_enabled=False, - vad_enabled=True, - vad_analyzer=SileroVADAnalyzer(), - transcription_enabled=True, - ) + transport = DailyTransport( + room_url, + token, + "Chatbot", + DailyParams( + api_url=daily_api_url, + api_key=daily_api_key, + audio_in_enabled=True, + audio_out_enabled=True, + camera_out_enabled=False, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + transcription_enabled=True, ) + ) - tts = ElevenLabsTTSService( - aiohttp_session=session, - api_key=os.getenv("ELEVENLABS_API_KEY", ""), - voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""), - ) + tts = ElevenLabsTTSService( + api_key=os.getenv("ELEVENLABS_API_KEY", ""), + voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""), + ) - llm = OpenAILLMService( - api_key=os.getenv("OPENAI_API_KEY"), - model="gpt-4o") + llm = OpenAILLMService( + api_key=os.getenv("OPENAI_API_KEY"), + model="gpt-4o") - messages = [ - { - "role": "system", - "content": "You are Chatbot, a friendly, helpful robot. Your output will be converted to audio so don't include special characters other than '!' or '?' in your answers. Respond to what the user said in a creative and helpful way, but keep your responses brief. Start by saying hello.", - }, - ] + messages = [ + { + "role": "system", + "content": "You are Chatbot, a friendly, helpful robot. Your output will be converted to audio so don't include special characters other than '!' or '?' in your answers. Respond to what the user said in a creative and helpful way, but keep your responses brief. Start by saying hello.", + }, + ] - tma_in = LLMUserResponseAggregator(messages) - tma_out = LLMAssistantResponseAggregator(messages) + tma_in = LLMUserResponseAggregator(messages) + tma_out = LLMAssistantResponseAggregator(messages) - pipeline = Pipeline([ - transport.input(), - tma_in, - llm, - tts, - transport.output(), - tma_out, - ]) + pipeline = Pipeline([ + transport.input(), + tma_in, + llm, + tts, + transport.output(), + tma_out, + ]) - task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) + task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) - @transport.event_handler("on_first_participant_joined") - async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) - await task.queue_frames([LLMMessagesFrame(messages)]) + @transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + transport.capture_participant_transcription(participant["id"]) + await task.queue_frames([LLMMessagesFrame(messages)]) - @transport.event_handler("on_participant_left") - async def on_participant_left(transport, participant, reason): + @transport.event_handler("on_participant_left") + async def on_participant_left(transport, participant, reason): + await task.queue_frame(EndFrame()) + + @transport.event_handler("on_call_state_updated") + async def on_call_state_updated(transport, state): + if state == "left": await task.queue_frame(EndFrame()) - @transport.event_handler("on_call_state_updated") - async def on_call_state_updated(transport, state): - if state == "left": - await task.queue_frame(EndFrame()) + runner = PipelineRunner() - runner = PipelineRunner() - - await runner.run(task) + await runner.run(task) if __name__ == "__main__": diff --git a/examples/dialin-chatbot/bot_daily.py b/examples/dialin-chatbot/bot_daily.py index ea30cd2d5..cd6afdad0 100644 --- a/examples/dialin-chatbot/bot_daily.py +++ b/examples/dialin-chatbot/bot_daily.py @@ -1,5 +1,4 @@ import asyncio -import aiohttp import os import sys import argparse @@ -29,75 +28,74 @@ daily_api_url = os.getenv("DAILY_API_URL", "https://api.daily.co/v1") async def main(room_url: str, token: str, callId: str, callDomain: str): - async with aiohttp.ClientSession() as session: - # diallin_settings are only needed if Daily's SIP URI is used - # If you are handling this via Twilio, Telnyx, set this to None - # and handle call-forwarding when on_dialin_ready fires. - diallin_settings = DailyDialinSettings( - call_id=callId, - call_domain=callDomain + # diallin_settings are only needed if Daily's SIP URI is used + # If you are handling this via Twilio, Telnyx, set this to None + # and handle call-forwarding when on_dialin_ready fires. + diallin_settings = DailyDialinSettings( + call_id=callId, + call_domain=callDomain + ) + + transport = DailyTransport( + room_url, + token, + "Chatbot", + DailyParams( + api_url=daily_api_url, + api_key=daily_api_key, + dialin_settings=diallin_settings, + audio_in_enabled=True, + audio_out_enabled=True, + camera_out_enabled=False, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + transcription_enabled=True, ) + ) - transport = DailyTransport( - room_url, - token, - "Chatbot", - DailyParams( - api_url=daily_api_url, - api_key=daily_api_key, - dialin_settings=diallin_settings, - audio_in_enabled=True, - audio_out_enabled=True, - camera_out_enabled=False, - vad_enabled=True, - vad_analyzer=SileroVADAnalyzer(), - transcription_enabled=True, - ) - ) + tts = ElevenLabsTTSService( + api_key=os.getenv("ELEVENLABS_API_KEY", ""), + voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""), + ) - tts = ElevenLabsTTSService( - aiohttp_session=session, - api_key=os.getenv("ELEVENLABS_API_KEY", ""), - voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""), - ) + llm = OpenAILLMService( + api_key=os.getenv("OPENAI_API_KEY"), + model="gpt-4o" + ) - llm = OpenAILLMService( - api_key=os.getenv("OPENAI_API_KEY"), - model="gpt-4o") + messages = [ + { + "role": "system", + "content": "You are Chatbot, a friendly, helpful robot. 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, but keep your responses brief. Start by saying 'Oh, hello! Who dares dial me at this hour?!'.", + }, + ] - messages = [ - { - "role": "system", - "content": "You are Chatbot, a friendly, helpful robot. 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, but keep your responses brief. Start by saying 'Oh, hello! Who dares dial me at this hour?!'.", - }, - ] + tma_in = LLMUserResponseAggregator(messages) + tma_out = LLMAssistantResponseAggregator(messages) - tma_in = LLMUserResponseAggregator(messages) - tma_out = LLMAssistantResponseAggregator(messages) + pipeline = Pipeline([ + transport.input(), + tma_in, + llm, + tts, + transport.output(), + tma_out, + ]) - pipeline = Pipeline([ - transport.input(), - tma_in, - llm, - tts, - transport.output(), - tma_out, - ]) + task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) - task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) + @transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + transport.capture_participant_transcription(participant["id"]) + await task.queue_frames([LLMMessagesFrame(messages)]) - @transport.event_handler("on_first_participant_joined") - async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) - await task.queue_frames([LLMMessagesFrame(messages)]) + @transport.event_handler("on_participant_left") + async def on_participant_left(transport, participant, reason): + await task.queue_frame(EndFrame()) - @transport.event_handler("on_participant_left") - async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + runner = PipelineRunner() - runner = PipelineRunner() - - await runner.run(task) + await runner.run(task) if __name__ == "__main__": diff --git a/examples/dialin-chatbot/bot_twilio.py b/examples/dialin-chatbot/bot_twilio.py index 6ae8a24b3..e6653babd 100644 --- a/examples/dialin-chatbot/bot_twilio.py +++ b/examples/dialin-chatbot/bot_twilio.py @@ -1,5 +1,4 @@ import asyncio -import aiohttp import os import sys import argparse @@ -36,82 +35,81 @@ daily_api_key = os.getenv("DAILY_API_KEY", "") async def main(room_url: str, token: str, callId: str, sipUri: str): - async with aiohttp.ClientSession() as session: - # diallin_settings are only needed if Daily's SIP URI is used - # If you are handling this via Twilio, Telnyx, set this to None - # and handle call-forwarding when on_dialin_ready fires. - transport = DailyTransport( - room_url, - token, - "Chatbot", - DailyParams( - api_key=daily_api_key, - dialin_settings=None, # Not required for Twilio - audio_in_enabled=True, - audio_out_enabled=True, - camera_out_enabled=False, - vad_enabled=True, - vad_analyzer=SileroVADAnalyzer(), - transcription_enabled=True, + # dialin_settings are only needed if Daily's SIP URI is used + # If you are handling this via Twilio, Telnyx, set this to None + # and handle call-forwarding when on_dialin_ready fires. + transport = DailyTransport( + room_url, + token, + "Chatbot", + DailyParams( + api_key=daily_api_key, + dialin_settings=None, # Not required for Twilio + audio_in_enabled=True, + audio_out_enabled=True, + camera_out_enabled=False, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + transcription_enabled=True, + ) + ) + + tts = ElevenLabsTTSService( + api_key=os.getenv("ELEVENLABS_API_KEY", ""), + voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""), + ) + + llm = OpenAILLMService( + api_key=os.getenv("OPENAI_API_KEY"), + model="gpt-4o" + ) + + messages = [ + { + "role": "system", + "content": "You are Chatbot, a friendly, helpful robot. 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, but keep your responses brief. Start by saying 'Hello! Who dares dial me at this hour?!'.", + }, + ] + + tma_in = LLMUserResponseAggregator(messages) + tma_out = LLMAssistantResponseAggregator(messages) + + pipeline = Pipeline([ + transport.input(), + tma_in, + llm, + tts, + transport.output(), + tma_out, + ]) + + task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) + + @transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + transport.capture_participant_transcription(participant["id"]) + await task.queue_frames([LLMMessagesFrame(messages)]) + + @transport.event_handler("on_participant_left") + async def on_participant_left(transport, participant, reason): + await task.queue_frame(EndFrame()) + + @transport.event_handler("on_dialin_ready") + async def on_dialin_ready(transport, cdata): + # For Twilio, Telnyx, etc. You need to update the state of the call + # and forward it to the sip_uri.. + print(f"Forwarding call: {callId} {sipUri}") + + try: + # The TwiML is updated using Twilio's client library + call = twilioclient.calls(callId).update( + twiml=f'{sipUri}' ) - ) + except Exception as e: + raise Exception(f"Failed to forward call: {str(e)}") - tts = ElevenLabsTTSService( - aiohttp_session=session, - api_key=os.getenv("ELEVENLABS_API_KEY", ""), - voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""), - ) - - llm = OpenAILLMService( - api_key=os.getenv("OPENAI_API_KEY"), - model="gpt-4o") - - messages = [ - { - "role": "system", - "content": "You are Chatbot, a friendly, helpful robot. 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, but keep your responses brief. Start by saying 'Hello! Who dares dial me at this hour?!'.", - }, - ] - - tma_in = LLMUserResponseAggregator(messages) - tma_out = LLMAssistantResponseAggregator(messages) - - pipeline = Pipeline([ - transport.input(), - tma_in, - llm, - tts, - transport.output(), - tma_out, - ]) - - task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) - - @transport.event_handler("on_first_participant_joined") - async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) - await task.queue_frames([LLMMessagesFrame(messages)]) - - @transport.event_handler("on_participant_left") - async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) - - @transport.event_handler("on_dialin_ready") - async def on_dialin_ready(transport, cdata): - # For Twilio, Telnyx, etc. You need to update the state of the call - # and forward it to the sip_uri.. - print(f"Forwarding call: {callId} {sipUri}") - - try: - # The TwiML is updated using Twilio's client library - call = twilioclient.calls(callId).update( - twiml=f'{sipUri}' - ) - except Exception as e: - raise Exception(f"Failed to forward call: {str(e)}") - - runner = PipelineRunner() - await runner.run(task) + runner = PipelineRunner() + await runner.run(task) if __name__ == "__main__": diff --git a/examples/foundational/05-sync-speech-and-image.py b/examples/foundational/05-sync-speech-and-image.py index e3965e857..ca3ff9557 100644 --- a/examples/foundational/05-sync-speech-and-image.py +++ b/examples/foundational/05-sync-speech-and-image.py @@ -89,7 +89,6 @@ async def main(): ) tts = ElevenLabsTTSService( - aiohttp_session=session, api_key=os.getenv("ELEVENLABS_API_KEY"), voice_id=os.getenv("ELEVENLABS_VOICE_ID"), ) diff --git a/examples/foundational/05a-local-sync-speech-and-image.py b/examples/foundational/05a-local-sync-speech-and-image.py index 5decffcb5..63bcf1e9d 100644 --- a/examples/foundational/05a-local-sync-speech-and-image.py +++ b/examples/foundational/05a-local-sync-speech-and-image.py @@ -85,7 +85,6 @@ async def main(): model="gpt-4o") tts = ElevenLabsTTSService( - aiohttp_session=session, api_key=os.getenv("ELEVENLABS_API_KEY"), voice_id=os.getenv("ELEVENLABS_VOICE_ID")) diff --git a/examples/foundational/06a-image-sync.py b/examples/foundational/06a-image-sync.py index fb1824ed8..812dab137 100644 --- a/examples/foundational/06a-image-sync.py +++ b/examples/foundational/06a-image-sync.py @@ -79,7 +79,6 @@ async def main(): ) tts = ElevenLabsTTSService( - aiohttp_session=session, api_key=os.getenv("ELEVENLABS_API_KEY"), voice_id=os.getenv("ELEVENLABS_VOICE_ID"), ) diff --git a/examples/foundational/07b-interruptible-langchain.py b/examples/foundational/07b-interruptible-langchain.py index c517ff27a..872dbf9bb 100644 --- a/examples/foundational/07b-interruptible-langchain.py +++ b/examples/foundational/07b-interruptible-langchain.py @@ -18,7 +18,6 @@ from pipecat.processors.aggregators.llm_response import ( LLMAssistantResponseAggregator, LLMUserResponseAggregator) from pipecat.processors.frameworks.langchain import LangchainProcessor from pipecat.services.cartesia import CartesiaTTSService -from pipecat.services.elevenlabs import ElevenLabsTTSService from pipecat.transports.services.daily import DailyParams, DailyTransport from pipecat.vad.silero import SileroVADAnalyzer diff --git a/examples/foundational/11-sound-effects.py b/examples/foundational/11-sound-effects.py index 00fb0c9be..146b3bd09 100644 --- a/examples/foundational/11-sound-effects.py +++ b/examples/foundational/11-sound-effects.py @@ -104,7 +104,6 @@ async def main(): model="gpt-4o") tts = ElevenLabsTTSService( - aiohttp_session=session, api_key=os.getenv("ELEVENLABS_API_KEY"), voice_id="ErXwobaYiN019PkySvjV", ) diff --git a/examples/simple-chatbot/bot.py b/examples/simple-chatbot/bot.py index d00f6acd1..1664e47fb 100644 --- a/examples/simple-chatbot/bot.py +++ b/examples/simple-chatbot/bot.py @@ -111,7 +111,6 @@ async def main(): ) tts = ElevenLabsTTSService( - aiohttp_session=session, api_key=os.getenv("ELEVENLABS_API_KEY"), # # English diff --git a/examples/storytelling-chatbot/src/bot.py b/examples/storytelling-chatbot/src/bot.py index 4bd50fe42..91452dd75 100644 --- a/examples/storytelling-chatbot/src/bot.py +++ b/examples/storytelling-chatbot/src/bot.py @@ -60,7 +60,6 @@ async def main(room_url, token=None): ) tts_service = ElevenLabsTTSService( - aiohttp_session=session, api_key=os.getenv("ELEVENLABS_API_KEY"), voice_id=os.getenv("ELEVENLABS_VOICE_ID"), ) diff --git a/pyproject.toml b/pyproject.toml index 721f4be19..73c643ddc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -39,6 +39,7 @@ azure = [ "azure-cognitiveservices-speech~=1.40.0" ] cartesia = [ "websockets~=12.0" ] daily = [ "daily-python~=0.10.1" ] deepgram = [ "deepgram-sdk~=3.5.0" ] +elevenlabs = [ "elevenlabs~=1.7.0" ] examples = [ "python-dotenv~=1.0.1", "flask~=3.0.3", "flask_cors~=4.0.1" ] fal = [ "fal-client~=0.4.1" ] gladia = [ "websockets~=12.0" ] diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index e3541ccea..429e7b3e4 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -10,7 +10,7 @@ import base64 import asyncio import time -from typing import AsyncGenerator, Mapping +from typing import AsyncGenerator from pipecat.frames.frames import ( CancelFrame, diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index 974619ea8..ed8041fcf 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -4,16 +4,36 @@ # SPDX-License-Identifier: BSD 2-Clause License # -import aiohttp - from typing import AsyncGenerator, Literal from pydantic import BaseModel -from pipecat.frames.frames import AudioRawFrame, ErrorFrame, Frame, TTSStartedFrame, TTSStoppedFrame +from pipecat.frames.frames import AudioRawFrame, Frame, TTSStartedFrame, TTSStoppedFrame from pipecat.services.ai_services import TTSService from loguru import logger +# See .env.example for ElevenLabs configuration needed +try: + from elevenlabs.client import AsyncElevenLabs +except ModuleNotFoundError as e: + logger.error(f"Exception: {e}") + logger.error( + "In order to use ElevenLabs, you need to `pip install pipecat-ai[elevenlabs]`. Also, set `ELEVENLABS_API_KEY` environment variable.") + raise Exception(f"Missing module: {e}") + + +def sample_rate_from_output_format(output_format: str) -> int: + match output_format: + case "pcm_16000": + return 16000 + case "pcm_22050": + return 22050 + case "pcm_24000": + return 24000 + case "pcm_44100": + return 44100 + return 16000 + class ElevenLabsTTSService(TTSService): class InputParams(BaseModel): @@ -24,21 +44,24 @@ class ElevenLabsTTSService(TTSService): *, api_key: str, voice_id: str, - aiohttp_session: aiohttp.ClientSession, model: str = "eleven_turbo_v2_5", params: InputParams = InputParams(), **kwargs): super().__init__(**kwargs) - self._api_key = api_key self._voice_id = voice_id self._model = model self._params = params - self._aiohttp_session = aiohttp_session + self._client = AsyncElevenLabs(api_key=api_key) + self._sample_rate = sample_rate_from_output_format(params.output_format) def can_generate_metrics(self) -> bool: return True + async def set_model(self, model: str): + logger.debug(f"Switching TTS model to: [{model}]") + self._model = model + async def set_voice(self, voice: str): logger.debug(f"Switching TTS voice to: [{voice}]") self._voice_id = voice @@ -46,34 +69,25 @@ class ElevenLabsTTSService(TTSService): async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: logger.debug(f"Generating TTS: [{text}]") - url = f"https://api.elevenlabs.io/v1/text-to-speech/{self._voice_id}/stream" - - payload = {"text": text, "model_id": self._model} - - querystring = { - "output_format": self._params.output_format - } - - headers = { - "xi-api-key": self._api_key, - "Content-Type": "application/json", - } - + await self.start_tts_usage_metrics(text) await self.start_ttfb_metrics() - async with self._aiohttp_session.post(url, json=payload, headers=headers, params=querystring) as r: - if r.status != 200: - text = await r.text() - logger.error(f"{self} error getting audio (status: {r.status}, error: {text})") - yield ErrorFrame(f"Error getting audio (status: {r.status}, error: {text})") - return + results = await self._client.generate( + text=text, + voice=self._voice_id, + model=self._model, + output_format=self._params.output_format + ) - await self.start_tts_usage_metrics(text) + tts_started = False + async for audio in results: + # This is so we send TTSStartedFrame when we have the first audio + # bytes. + if not tts_started: + await self.push_frame(TTSStartedFrame()) + tts_started = True + await self.stop_ttfb_metrics() + frame = AudioRawFrame(audio, self._sample_rate, 1) + yield frame - await self.push_frame(TTSStartedFrame()) - async for chunk in r.content: - if len(chunk) > 0: - await self.stop_ttfb_metrics() - frame = AudioRawFrame(chunk, 16000, 1) - yield frame - await self.push_frame(TTSStoppedFrame()) + await self.push_frame(TTSStoppedFrame()) From 748a7af6029eeb00111e5c97a57b83933f14cbfb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 5 Sep 2024 19:05:29 -0700 Subject: [PATCH 04/25] update CHANGELOG.md --- CHANGELOG.md | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7bf00326d..96fe57619 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Interruptions support has been added to `TwilioFrameSerializer` when using + `FastAPIWebsocketTransport`. + - Added new `LmntTTSService` text-to-speech service. (see https://www.lmnt.com/) @@ -32,6 +35,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 big chunk (i.e. from when the user starts speaking until the user stops speaking) instead of a continous stream. +### Fixed + +- `StartFrame` should be the first frame every processor receives to avoid + situations where things are not initialized (because initialization happens on + `StartFrame`) and other frames come in resulting in undesired behavior. + +### Performance + +- `obj_id()` and `obj_count()` now use `itertools.count` avoiding the need of + `threading.Lock`. + ## [0.0.41] - 2024-08-22 ### Added From c92531a02f5433835c02bd69eecab9964bb91c86 Mon Sep 17 00:00:00 2001 From: "Danny D. Leybzon" Date: Fri, 6 Sep 2024 20:22:18 +0200 Subject: [PATCH 05/25] Update requirements.txt request.form() throws an error if you don't have python-multipart installed --- examples/dialin-chatbot/requirements.txt | 1 + 1 file changed, 1 insertion(+) diff --git a/examples/dialin-chatbot/requirements.txt b/examples/dialin-chatbot/requirements.txt index 38a0a93b0..e59a9c3d2 100644 --- a/examples/dialin-chatbot/requirements.txt +++ b/examples/dialin-chatbot/requirements.txt @@ -3,3 +3,4 @@ fastapi uvicorn python-dotenv twilio +python-multipart From 081b001c8b8af41dfd57fb779b88e21fb843dca6 Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Sat, 7 Sep 2024 16:42:52 -0700 Subject: [PATCH 06/25] fix for leading space in context aggregator strings --- src/pipecat/processors/aggregators/llm_response.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 7c38e62ad..ab0552578 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -109,7 +109,7 @@ class LLMResponseAggregator(FrameProcessor): await self.push_frame(frame, direction) elif isinstance(frame, self._accumulator_frame): if self._aggregating: - self._aggregation += f" {frame.text}" + self._aggregation += f" {frame.text}" if self._aggregation else frame.text # We have recevied a complete sentence, so if we have seen the # end frame and we were still aggregating, it means we should # send the aggregation. From 98286336bf46b8592b2f4e598ab1a62cf5a1e32a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Sun, 8 Sep 2024 19:38:16 -0700 Subject: [PATCH 07/25] transports(daily): allow setting audio output bitrate (default 96kpbs) Fixes #388 --- CHANGELOG.md | 7 +++++++ src/pipecat/transports/base_transport.py | 1 + src/pipecat/transports/services/daily.py | 6 ++++++ 3 files changed, 14 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 96fe57619..9badb6c70 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 +- `DailyTransport` now supports setting the audio bitrate to improve audio + quality through the `DailyParams.audio_out_bitrate` parameter. The new + default is 96kbps. + +- `DailyTransport` now uses the number of audio output channels (1 or 2) to set + mono or stereo audio when needed. + - Interruptions support has been added to `TwilioFrameSerializer` when using `FastAPIWebsocketTransport`. diff --git a/src/pipecat/transports/base_transport.py b/src/pipecat/transports/base_transport.py index 72e609263..083aeac37 100644 --- a/src/pipecat/transports/base_transport.py +++ b/src/pipecat/transports/base_transport.py @@ -32,6 +32,7 @@ class TransportParams(BaseModel): audio_out_is_live: bool = False audio_out_sample_rate: int = 16000 audio_out_channels: int = 1 + audio_out_bitrate: int = 96000 audio_in_enabled: bool = False audio_in_sample_rate: int = 16000 audio_in_channels: int = 1 diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index bb6032fa4..7cf330b9e 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -366,6 +366,12 @@ class DailyTransportClient(EventHandler): } }, } + }, + "microphone": { + "sendSettings": { + "channelConfig": "stereo" if self._params.audio_out_channels == 2 else "mono", + "bitrate": self._params.audio_out_bitrate, + } } }, }) From 3cbe97d346c42d5694ba06f35ec92d2c5feeab00 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 12 Sep 2024 00:31:23 -0700 Subject: [PATCH 08/25] clocks: added new BaseClock and SystemClock --- src/pipecat/clocks/__init__.py | 0 src/pipecat/clocks/base_clock.py | 18 ++++++++++++++++++ src/pipecat/clocks/system_clock.py | 21 +++++++++++++++++++++ src/pipecat/transcriptions/__init__.py | 0 4 files changed, 39 insertions(+) create mode 100644 src/pipecat/clocks/__init__.py create mode 100644 src/pipecat/clocks/base_clock.py create mode 100644 src/pipecat/clocks/system_clock.py create mode 100644 src/pipecat/transcriptions/__init__.py diff --git a/src/pipecat/clocks/__init__.py b/src/pipecat/clocks/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/pipecat/clocks/base_clock.py b/src/pipecat/clocks/base_clock.py new file mode 100644 index 000000000..aa7b7b806 --- /dev/null +++ b/src/pipecat/clocks/base_clock.py @@ -0,0 +1,18 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +from abc import ABC, abstractmethod + + +class BaseClock(ABC): + + @abstractmethod + def get_time(self) -> int: + pass + + @abstractmethod + def start(self): + pass diff --git a/src/pipecat/clocks/system_clock.py b/src/pipecat/clocks/system_clock.py new file mode 100644 index 000000000..20319cff6 --- /dev/null +++ b/src/pipecat/clocks/system_clock.py @@ -0,0 +1,21 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import time + +from pipecat.clocks.base_clock import BaseClock + + +class SystemClock(BaseClock): + + def __init__(self): + self._time = 0 + + def get_time(self) -> int: + return time.monotonic_ns() - self._time if self._time > 0 else 0 + + def start(self): + self._time = time.monotonic_ns() diff --git a/src/pipecat/transcriptions/__init__.py b/src/pipecat/transcriptions/__init__.py new file mode 100644 index 000000000..e69de29bb From 72f231b32732ab572c35291262c67e903815efbb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 12 Sep 2024 00:17:04 -0700 Subject: [PATCH 09/25] frames: add a presentation timestamp (pts) to each frame --- src/pipecat/frames/frames.py | 58 ++++++++++++++++++++++++++++++------ src/pipecat/utils/time.py | 17 +++++++++++ 2 files changed, 66 insertions(+), 9 deletions(-) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 13c2f53f1..9cad1268d 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -8,19 +8,27 @@ from typing import Any, List, Mapping, Optional, Tuple from dataclasses import dataclass, field +from pipecat.clocks.base_clock import BaseClock from pipecat.transcriptions.language import Language +from pipecat.utils.time import nanoseconds_to_str from pipecat.utils.utils import obj_count, obj_id from pipecat.vad.vad_analyzer import VADParams +def format_pts(pts: int | None): + return nanoseconds_to_str(pts) if pts else None + + @dataclass class Frame: id: int = field(init=False) name: str = field(init=False) + pts: Optional[int] = field(init=False) def __post_init__(self): self.id: int = obj_id() self.name: str = f"{self.__class__.__name__}#{obj_count(self)}" + self.pts: Optional[int] = None def __str__(self): return self.name @@ -46,7 +54,14 @@ class AudioRawFrame(DataFrame): self.num_frames = int(len(self.audio) / (self.num_channels * 2)) def __str__(self): - return f"{self.name}(size: {len(self.audio)}, frames: {self.num_frames}, sample_rate: {self.sample_rate}, channels: {self.num_channels})" + pts = format_pts(self.pts) + return f"{ + self.name}(pts: {pts}, size: { + len( + self.audio)}, frames: { + self.num_frames}, sample_rate: { + self.sample_rate}, channels: { + self.num_channels})" @dataclass @@ -60,7 +75,8 @@ class ImageRawFrame(DataFrame): format: str | None def __str__(self): - return f"{self.name}(size: {self.size}, format: {self.format})" + pts = format_pts(self.pts) + return f"{self.name}(pts: {pts}, size: {self.size}, format: {self.format})" @dataclass @@ -72,7 +88,8 @@ class URLImageRawFrame(ImageRawFrame): url: str | None def __str__(self): - return f"{self.name}(url: {self.url}, size: {self.size}, format: {self.format})" + pts = format_pts(self.pts) + return f"{self.name}(pts: {pts}, url: {self.url}, size: {self.size}, format: {self.format})" @dataclass @@ -84,7 +101,12 @@ class VisionImageRawFrame(ImageRawFrame): text: str | None def __str__(self): - return f"{self.name}(text: {self.text}, size: {self.size}, format: {self.format})" + pts = format_pts(self.pts) + return f"{ + self.name}(pts: {pts}, text: { + self.text}, size: { + self.size}, format: { + self.format})" @dataclass @@ -96,7 +118,12 @@ class UserImageRawFrame(ImageRawFrame): user_id: str def __str__(self): - return f"{self.name}(user: {self.user_id}, size: {self.size}, format: {self.format})" + pts = format_pts(self.pts) + return f"{ + self.name}(pts: {pts}, user: { + self.user_id}, size: { + self.size}, format: { + self.format})" @dataclass @@ -109,7 +136,8 @@ class SpriteFrame(Frame): images: List[ImageRawFrame] def __str__(self): - return f"{self.name}(size: {len(self.images)})" + pts = format_pts(self.pts) + return f"{self.name}(pts: {pts}, size: {len(self.images)})" @dataclass @@ -121,7 +149,8 @@ class TextFrame(DataFrame): text: str def __str__(self): - return f"{self.name}(text: {self.text})" + pts = format_pts(self.pts) + return f"{self.name}(pts: {pts}, text: {self.text})" @dataclass @@ -135,7 +164,12 @@ class TranscriptionFrame(TextFrame): language: Language | None = None def __str__(self): - return f"{self.name}(user: {self.user_id}, text: {self.text}, language: {self.language}, timestamp: {self.timestamp})" + return f"{ + self.name}(user: { + self.user_id}, text: { + self.text}, language: { + self.language}, timestamp: { + self.timestamp})" @dataclass @@ -147,7 +181,12 @@ class InterimTranscriptionFrame(TextFrame): language: Language | None = None def __str__(self): - return f"{self.name}(user: {self.user_id}, text: {self.text}, language: {self.language}, timestamp: {self.timestamp})" + return f"{ + self.name}(user: { + self.user_id}, text: { + self.text}, language: { + self.language}, timestamp: { + self.timestamp})" @dataclass @@ -326,6 +365,7 @@ class ControlFrame(Frame): @dataclass class StartFrame(ControlFrame): """This is the first frame that should be pushed down a pipeline.""" + clock: BaseClock allow_interruptions: bool = False enable_metrics: bool = False enable_usage_metrics: bool = False diff --git a/src/pipecat/utils/time.py b/src/pipecat/utils/time.py index af493e77b..0f6ca1076 100644 --- a/src/pipecat/utils/time.py +++ b/src/pipecat/utils/time.py @@ -9,3 +9,20 @@ import datetime def time_now_iso8601() -> str: return datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="milliseconds") + + +def seconds_to_nanoseconds(seconds: float) -> int: + return int(seconds * 1_000_000_000) + + +def nanoseconds_to_seconds(nanoseconds: int) -> float: + return nanoseconds / 1_000_000_000 + + +def nanoseconds_to_str(nanoseconds: int) -> str: + total_seconds = nanoseconds_to_seconds(nanoseconds) + hours = int(total_seconds // 3600) + minutes = int((total_seconds % 3600) // 60) + seconds = int(total_seconds % 60) + microseconds = int((total_seconds - int(total_seconds)) * 1_000_000) + return f"{hours}:{minutes:02}:{seconds:02}.{microseconds:06}" From 7807cbeb397804ad0c500432d39b24ee10f21743 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 12 Sep 2024 00:18:24 -0700 Subject: [PATCH 10/25] pipeline(task): add a clock to the pipeline task --- src/pipecat/pipeline/task.py | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 102b4528b..03fd5c734 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -10,6 +10,8 @@ from typing import AsyncIterable, Iterable from pydantic import BaseModel +from pipecat.clocks.base_clock import BaseClock +from pipecat.clocks.system_clock import SystemClock from pipecat.frames.frames import ( CancelFrame, EndFrame, @@ -60,11 +62,16 @@ class Source(FrameProcessor): class PipelineTask: - def __init__(self, pipeline: BasePipeline, params: PipelineParams = PipelineParams()): + def __init__( + self, + pipeline: BasePipeline, + params: PipelineParams = PipelineParams(), + clock: BaseClock = SystemClock()): self.id: int = obj_id() self.name: str = f"{self.__class__.__name__}#{obj_count(self)}" self._pipeline = pipeline + self._clock = clock self._params = params self._finished = False @@ -116,11 +123,14 @@ class PipelineTask: return MetricsFrame(ttfb=ttfb, processing=processing) async def _process_down_queue(self): + self._clock.start() + start_frame = StartFrame( allow_interruptions=self._params.allow_interruptions, enable_metrics=self._params.enable_metrics, enable_usage_metrics=self._params.enable_metrics, - report_only_initial_ttfb=self._params.report_only_initial_ttfb + report_only_initial_ttfb=self._params.report_only_initial_ttfb, + clock=self._clock ) await self._source.process_frame(start_frame, FrameDirection.DOWNSTREAM) From 7749692f72181916725bfa075002bfb6b3f8cd59 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 12 Sep 2024 00:18:29 -0700 Subject: [PATCH 11/25] processors: get pipeline clock from StartFrame --- src/pipecat/processors/frame_processor.py | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 156e1c0ae..e228ec6b4 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -9,6 +9,7 @@ import time from enum import Enum +from pipecat.clocks.base_clock import BaseClock from pipecat.frames.frames import ( ErrorFrame, Frame, @@ -69,7 +70,10 @@ class FrameProcessorMetrics: async def start_llm_usage_metrics(self, tokens: dict): logger.debug( - f"{self._name} prompt tokens: {tokens['prompt_tokens']}, completion tokens: {tokens['completion_tokens']}") + f"{ + self._name} prompt tokens: { + tokens['prompt_tokens']}, completion tokens: { + tokens['completion_tokens']}") return MetricsFrame(tokens=[tokens]) async def start_tts_usage_metrics(self, text: str): @@ -96,6 +100,9 @@ class FrameProcessor: self._next: "FrameProcessor" | None = None self._loop: asyncio.AbstractEventLoop = loop or asyncio.get_running_loop() + # Clock + self._clock: BaseClock | None = None + # Properties self._allow_interruptions = False self._enable_metrics = False @@ -177,8 +184,12 @@ class FrameProcessor: def get_parent(self) -> "FrameProcessor": return self._parent + def get_clock(self) -> BaseClock: + return self._clock + async def process_frame(self, frame: Frame, direction: FrameDirection): if isinstance(frame, StartFrame): + self._clock = frame.clock self._allow_interruptions = frame.allow_interruptions self._enable_metrics = frame.enable_metrics self._enable_usage_metrics = frame.enable_usage_metrics From 02d926e9bd75dd011cad844645087a0365842441 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 12 Sep 2024 00:18:55 -0700 Subject: [PATCH 12/25] services: create AsyncTTSService and AsyncWordTTSService --- src/pipecat/services/ai_services.py | 74 ++++++++++++++++++++++++++++- 1 file changed, 73 insertions(+), 1 deletion(-) diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index c5cb7cc0d..72a8828c2 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -6,10 +6,11 @@ import asyncio import io +import time import wave from abc import abstractmethod -from typing import AsyncGenerator, Optional +from typing import AsyncGenerator, List, Optional, Tuple from pipecat.frames.frames import ( AudioRawFrame, @@ -37,9 +38,12 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.transcriptions.language import Language from pipecat.utils.audio import calculate_audio_volume from pipecat.utils.string import match_endofsentence +from pipecat.utils.time import seconds_to_nanoseconds from pipecat.utils.utils import exp_smoothing from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from loguru import logger + class AIService(FrameProcessor): def __init__(self, **kwargs): @@ -303,6 +307,74 @@ class TTSService(AIService): pass +class AsyncTTSService(TTSService): + def __init__(self, **kwargs): + super().__init__(**kwargs) + + @abstractmethod + async def flush_audio(self): + pass + + +class AsyncWordTTSService(AsyncTTSService): + def __init__(self, **kwargs): + super().__init__(**kwargs) + self._start_word_timestamp = None + self._words_queue = asyncio.Queue() + self._words_task = self.get_event_loop().create_task(self._words_task_handler()) + + def init_word_timestamps(self): + if not self._start_word_timestamp: + self._start_word_timestamp = self.get_clock().get_time() + + def reset_word_timestamps(self): + self._start_word_timestamp = None + self._word_timestamps = [] + + async def add_word_timestamps(self, word_times: List[Tuple[str, float]]): + for (word, timestamp) in word_times: + await self._words_queue.put((word, seconds_to_nanoseconds(timestamp))) + + async def stop(self, frame: EndFrame): + await super().stop(frame) + await self._stop_words_task() + + async def cancel(self, frame: CancelFrame): + await super().cancel(frame) + await self._stop_words_task() + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if isinstance(frame, LLMFullResponseEndFrame) or isinstance(frame, EndFrame): + await self.flush_audio() + + async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection): + await super()._handle_interruption(frame, direction) + self.reset_word_timestamps() + + async def _stop_words_task(self): + if self._words_task: + self._words_task.cancel() + await self._words_task + + async def _words_task_handler(self): + while True: + try: + (word, timestamp) = await self._words_queue.get() + if word == "LLMFullResponseEndFrame" and timestamp == 0: + await self.push_frame(LLMFullResponseEndFrame()) + else: + frame = TextFrame(word) + frame.pts = self._start_word_timestamp + timestamp + await self.push_frame(frame) + self._words_queue.task_done() + except asyncio.CancelledError: + break + except Exception as e: + logger.exception(f"{self} exception: {e}") + + class STTService(AIService): """STTService is a base class for speech-to-text services.""" From 80f6d74e808e4b7ddeb347aed792eda47cfa14d0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 12 Sep 2024 00:19:37 -0700 Subject: [PATCH 13/25] services(cartesia): change to subclass of AsyncWordTTSService --- src/pipecat/services/cartesia.py | 69 ++++++++------------------------ 1 file changed, 16 insertions(+), 53 deletions(-) diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index 429e7b3e4..f263db60d 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -27,7 +27,7 @@ from pipecat.frames.frames import ( ) from pipecat.processors.frame_processor import FrameDirection from pipecat.transcriptions.language import Language -from pipecat.services.ai_services import TTSService +from pipecat.services.ai_services import AsyncWordTTSService from loguru import logger @@ -60,7 +60,7 @@ def language_to_cartesia_language(language: Language) -> str | None: return None -class CartesiaTTSService(TTSService): +class CartesiaTTSService(AsyncWordTTSService): def __init__( self, @@ -74,19 +74,17 @@ class CartesiaTTSService(TTSService): sample_rate: int = 16000, language: str = "en", **kwargs): - super().__init__(**kwargs) - # Aggregating sentences still gives cleaner-sounding results and fewer - # artifacts than streaming one word at a time. On average, waiting for - # a full sentence should only "cost" us 15ms or so with GPT-4o or a Llama 3 - # model, and it's worth it for the better audio quality. - self._aggregate_sentences = True - - # we don't want to automatically push LLM response text frames, because the - # context aggregators will add them to the LLM context even if we're - # interrupted. cartesia gives us word-by-word timestamps. we can use those - # to generate text frames ourselves aligned with the playout timing of the audio! - self._push_text_frames = False + # artifacts than streaming one word at a time. On average, waiting for a + # full sentence should only "cost" us 15ms or so with GPT-4o or a Llama + # 3 model, and it's worth it for the better audio quality. + # + # We also don't want to automatically push LLM response text frames, + # because the context aggregators will add them to the LLM context even + # if we're interrupted. Cartesia gives us word-by-word timestamps. We + # can use those to generate text frames ourselves aligned with the + # playout timing of the audio! + super().__init__(aggregate_sentences=True, push_text_frames=False, **kwargs) self._api_key = api_key self._cartesia_version = cartesia_version @@ -102,10 +100,7 @@ class CartesiaTTSService(TTSService): self._websocket = None self._context_id = None - self._context_id_start_timestamp = None - self._timestamped_words_buffer = [] self._receive_task = None - self._context_appending_task = None def can_generate_metrics(self) -> bool: return True @@ -140,7 +135,6 @@ class CartesiaTTSService(TTSService): f"{self._url}?api_key={self._api_key}&cartesia_version={self._cartesia_version}" ) self._receive_task = self.get_event_loop().create_task(self._receive_task_handler()) - self._context_appending_task = self.get_event_loop().create_task(self._context_appending_task_handler()) except Exception as e: logger.exception(f"{self} initialization error: {e}") self._websocket = None @@ -149,10 +143,6 @@ class CartesiaTTSService(TTSService): try: await self.stop_all_metrics() - if self._context_appending_task: - self._context_appending_task.cancel() - await self._context_appending_task - self._context_appending_task = None if self._receive_task: self._receive_task.cancel() await self._receive_task @@ -162,18 +152,14 @@ class CartesiaTTSService(TTSService): self._websocket = None self._context_id = None - self._context_id_start_timestamp = None - self._timestamped_words_buffer = [] except Exception as e: logger.exception(f"{self} error closing websocket: {e}") async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection): await super()._handle_interruption(frame, direction) - self._context_id = None - self._context_id_start_timestamp = None - self._timestamped_words_buffer = [] await self.stop_all_metrics() await self.push_frame(LLMFullResponseEndFrame()) + self._context_id = None async def _receive_task_handler(self): try: @@ -188,16 +174,14 @@ class CartesiaTTSService(TTSService): # because we are likely still playing out audio and need the # timestamp to set send context frames. self._context_id = None - self._timestamped_words_buffer.append(("LLMFullResponseEndFrame", 0)) + await self.add_word_timestamps([("LLMFullResponseEndFrame", 0)]) elif msg["type"] == "timestamps": - # logger.debug(f"TIMESTAMPS: {msg}") - self._timestamped_words_buffer.extend( + await self.add_word_timestamps( list(zip(msg["word_timestamps"]["words"], msg["word_timestamps"]["end"])) ) elif msg["type"] == "chunk": await self.stop_ttfb_metrics() - if not self._context_id_start_timestamp: - self._context_id_start_timestamp = time.time() + self.init_word_timestamps() frame = AudioRawFrame( audio=base64.b64decode(msg["data"]), sample_rate=self._output_format["sample_rate"], @@ -216,27 +200,6 @@ class CartesiaTTSService(TTSService): except Exception as e: logger.exception(f"{self} exception: {e}") - async def _context_appending_task_handler(self): - try: - while True: - await asyncio.sleep(0.1) - if not self._context_id_start_timestamp: - continue - elapsed_seconds = time.time() - self._context_id_start_timestamp - # Pop all words from self._timestamped_words_buffer that are - # older than the elapsed time and print a message about them to - # the console. - while self._timestamped_words_buffer and self._timestamped_words_buffer[0][1] <= elapsed_seconds: - word, timestamp = self._timestamped_words_buffer.pop(0) - if word == "LLMFullResponseEndFrame" and timestamp == 0: - await self.push_frame(LLMFullResponseEndFrame()) - continue - await self.push_frame(TextFrame(word)) - except asyncio.CancelledError: - pass - except Exception as e: - logger.exception(f"{self} exception: {e}") - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: logger.debug(f"Generating TTS: [{text}]") From a98d78cdea6402253f7c417a68c08d3656138f47 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 12 Sep 2024 00:20:06 -0700 Subject: [PATCH 14/25] services(lmnt): change to subclass of AsyncTTSService --- src/pipecat/services/lmnt.py | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/src/pipecat/services/lmnt.py b/src/pipecat/services/lmnt.py index f7afd6a41..59dd0aa5f 100644 --- a/src/pipecat/services/lmnt.py +++ b/src/pipecat/services/lmnt.py @@ -20,7 +20,7 @@ from pipecat.frames.frames import ( TTSStartedFrame, TTSStoppedFrame, ) -from pipecat.services.ai_services import TTSService +from pipecat.services.ai_services import AsyncTTSService from loguru import logger @@ -34,7 +34,7 @@ except ModuleNotFoundError as e: raise Exception(f"Missing module: {e}") -class LmntTTSService(TTSService): +class LmntTTSService(AsyncTTSService): def __init__( self, @@ -44,11 +44,9 @@ class LmntTTSService(TTSService): sample_rate: int = 24000, language: str = "en", **kwargs): - super().__init__(**kwargs) - # Let TTSService produce TTSStoppedFrames after a short delay of # no activity. - self._push_stop_frames = True + super().__init__(push_stop_frames=True, **kwargs) self._api_key = api_key self._voice_id = voice_id From 3665734972f525e0fafc22f86ad758b3c082212f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 12 Sep 2024 00:22:32 -0700 Subject: [PATCH 15/25] transports(output): initial sink clock synchronization --- src/pipecat/transports/base_output.py | 100 +++++++++++++++++++------- 1 file changed, 76 insertions(+), 24 deletions(-) diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index d2c6add2b..a24c9f4d2 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -8,6 +8,7 @@ import asyncio import itertools import time +import sys from PIL import Image from typing import List @@ -30,11 +31,14 @@ from pipecat.frames.frames import ( SystemFrame, TTSStartedFrame, TTSStoppedFrame, + TextFrame, TransportMessageFrame) from pipecat.transports.base_transport import TransportParams from loguru import logger +from pipecat.utils.time import nanoseconds_to_seconds + class BaseOutputTransport(FrameProcessor): @@ -64,7 +68,7 @@ class BaseOutputTransport(FrameProcessor): # Create sink frame task. This is the task that will actually write # audio or video frames. We write audio/video in a task so we can keep # generating frames upstream while, for example, the audio is playing. - self._create_sink_task() + self._create_sink_tasks() # Create push frame task. This is the task that will push frames in # order. We also guarantee that all frames are pushed in the same task. @@ -149,6 +153,7 @@ class BaseOutputTransport(FrameProcessor): await self._sink_queue.put(frame) await self.start(frame) elif isinstance(frame, EndFrame): + await self._sink_clock_queue.put((sys.maxsize, frame.id, frame)) await self._sink_queue.put(frame) await self.stop(frame) # Other frames. @@ -158,6 +163,9 @@ class BaseOutputTransport(FrameProcessor): await self._handle_image(frame) elif isinstance(frame, TransportMessageFrame) and frame.urgent: await self.send_message(frame) + # TODO(aleix): Images and audio should support presentation timestamps. + elif frame.pts: + await self._sink_clock_queue.put((frame.pts, frame.id, frame)) else: await self._sink_queue.put(frame) @@ -166,10 +174,14 @@ class BaseOutputTransport(FrameProcessor): return if isinstance(frame, StartInterruptionFrame): - # Stop sink task. + # Stop sink tasks. self._sink_task.cancel() await self._sink_task - self._create_sink_task() + # Stop sink clock tasks. + self._sink_clock_task.cancel() + await self._sink_clock_task + # Create sink tasks. + self._create_sink_tasks() # Stop push task. self._push_frame_task.cancel() await self._push_frame_task @@ -201,43 +213,83 @@ class BaseOutputTransport(FrameProcessor): else: await self._sink_queue.put(frame) - def _create_sink_task(self): + # + # Sink tasks + # + + def _create_sink_tasks(self): loop = self.get_event_loop() self._sink_queue = asyncio.Queue() self._sink_task = loop.create_task(self._sink_task_handler()) + self._sink_clock_queue = asyncio.PriorityQueue() + self._sink_clock_task = loop.create_task(self._sink_clock_task_handler()) + + async def _sink_frame_handler(self, frame: Frame): + if isinstance(frame, AudioRawFrame): + await self.write_raw_audio_frames(frame.audio) + await self._internal_push_frame(frame) + await self.push_frame(BotSpeakingFrame(), FrameDirection.UPSTREAM) + elif isinstance(frame, ImageRawFrame): + await self._set_camera_image(frame) + elif isinstance(frame, SpriteFrame): + await self._set_camera_images(frame.images) + elif isinstance(frame, TransportMessageFrame): + await self.send_message(frame) + elif isinstance(frame, TTSStartedFrame): + await self._bot_started_speaking() + await self._internal_push_frame(frame) + elif isinstance(frame, TTSStoppedFrame): + await self._bot_stopped_speaking() + await self._internal_push_frame(frame) + else: + await self._internal_push_frame(frame) async def _sink_task_handler(self): running = True while running: try: frame = await self._sink_queue.get() - if isinstance(frame, AudioRawFrame): - await self.write_raw_audio_frames(frame.audio) - await self._internal_push_frame(frame) - await self.push_frame(BotSpeakingFrame(), FrameDirection.UPSTREAM) - elif isinstance(frame, ImageRawFrame): - await self._set_camera_image(frame) - elif isinstance(frame, SpriteFrame): - await self._set_camera_images(frame.images) - elif isinstance(frame, TransportMessageFrame): - await self.send_message(frame) - elif isinstance(frame, TTSStartedFrame): - await self._bot_started_speaking() - await self._internal_push_frame(frame) - elif isinstance(frame, TTSStoppedFrame): - await self._bot_stopped_speaking() - await self._internal_push_frame(frame) - else: - await self._internal_push_frame(frame) - + await self._sink_frame_handler(frame) running = not isinstance(frame, EndFrame) - self._sink_queue.task_done() except asyncio.CancelledError: break except Exception as e: logger.exception(f"{self} error processing sink queue: {e}") + async def _sink_clock_frame_handler(self, frame: Frame): + # TODO(aleix): For now we just process TextFrame. But we should process + # audio and video as well. + if isinstance(frame, TextFrame): + await self._internal_push_frame(frame) + + async def _sink_clock_task_handler(self): + running = True + while running: + try: + timestamp, _, frame = await self._sink_clock_queue.get() + + # If we hit an EndFrame, we cna finish right away. + running = not isinstance(frame, EndFrame) + + # If we have a frame we check it's presentation timestamp. If it + # has already passed we process it, otherwise we wait until it's + # time to process it. + if running: + current_time = self.get_clock().get_time() + if timestamp <= current_time: + await self._sink_clock_frame_handler(frame) + else: + wait_time = nanoseconds_to_seconds(timestamp - current_time) + await asyncio.sleep(wait_time) + await self._sink_frame_handler(frame) + + self._sink_clock_queue.task_done() + except asyncio.CancelledError: + break + except Exception as e: + logger.exception(f"{self} error processing sink clock queue: {e}") + async def _bot_started_speaking(self): logger.debug("Bot started speaking") self._bot_speaking = True From f08b25dbb2132939c7c0cef107c3370a6a859dda Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 12 Sep 2024 00:29:43 -0700 Subject: [PATCH 16/25] examples: assistant aggregator should always goes after transport --- examples/foundational/07d-interruptible-cartesia.py | 2 +- examples/studypal/studypal.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/examples/foundational/07d-interruptible-cartesia.py b/examples/foundational/07d-interruptible-cartesia.py index 6b8bbcc5f..6bf5f9bd9 100644 --- a/examples/foundational/07d-interruptible-cartesia.py +++ b/examples/foundational/07d-interruptible-cartesia.py @@ -74,8 +74,8 @@ async def main(): tma_in, # User responses llm, # LLM tts, # TTS - tma_out, # Goes before the transport because cartesia has word-level timestamps! transport.output(), # Transport bot output + tma_out # Assistant spoken responses ]) task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True, enable_metrics=True)) diff --git a/examples/studypal/studypal.py b/examples/studypal/studypal.py index 8adfe2954..368a9b072 100644 --- a/examples/studypal/studypal.py +++ b/examples/studypal/studypal.py @@ -147,8 +147,8 @@ Your task is to help the user understand and learn from this article in 2 senten tma_in, llm, tts, - tma_out, transport.output(), + tma_out, ]) task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True, enable_metrics=True)) From 434493b8aad79c17e65d20f7acc7512aec77477b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 13 Sep 2024 09:31:35 -0700 Subject: [PATCH 17/25] services(elevenlabs): implement word-by-word support through websockets --- .../07d-interruptible-cartesia.py | 97 -------- pyproject.toml | 2 +- src/pipecat/services/ai_services.py | 15 +- src/pipecat/services/cartesia.py | 2 +- src/pipecat/services/elevenlabs.py | 232 ++++++++++++++++-- src/pipecat/services/lmnt.py | 2 + 6 files changed, 216 insertions(+), 134 deletions(-) delete mode 100644 examples/foundational/07d-interruptible-cartesia.py diff --git a/examples/foundational/07d-interruptible-cartesia.py b/examples/foundational/07d-interruptible-cartesia.py deleted file mode 100644 index 6bf5f9bd9..000000000 --- a/examples/foundational/07d-interruptible-cartesia.py +++ /dev/null @@ -1,97 +0,0 @@ -# -# Copyright (c) 2024, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -import aiohttp -import asyncio -import os -import sys - -from pipecat.frames.frames import LLMMessagesFrame -from pipecat.pipeline.pipeline import Pipeline -from pipecat.pipeline.runner import PipelineRunner -from pipecat.pipeline.task import PipelineParams, PipelineTask -from pipecat.processors.aggregators.llm_response import ( - LLMAssistantResponseAggregator, LLMUserResponseAggregator) -from pipecat.services.cartesia import CartesiaTTSService -from pipecat.services.openai import OpenAILLMService -from pipecat.transports.services.daily import DailyParams, DailyTransport -from pipecat.vad.silero import SileroVADAnalyzer - - -from runner import configure - -from loguru import logger - -from dotenv import load_dotenv -load_dotenv(override=True) - -logger.remove(0) -logger.add(sys.stderr, level="DEBUG") - - -async def main(): - async with aiohttp.ClientSession() as session: - (room_url, token) = await configure(session) - - transport = DailyTransport( - room_url, - token, - "Respond bot", - DailyParams( - audio_out_sample_rate=44100, - audio_out_enabled=True, - transcription_enabled=True, - vad_enabled=True, - vad_analyzer=SileroVADAnalyzer() - ) - ) - - tts = CartesiaTTSService( - api_key=os.getenv("CARTESIA_API_KEY"), - voice_id="a0e99841-438c-4a64-b679-ae501e7d6091", # Barbershop Man - sample_rate=44100, - ) - - llm = OpenAILLMService( - api_key=os.getenv("OPENAI_API_KEY"), - model="gpt-4o") - - 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.", - }, - ] - - tma_in = LLMUserResponseAggregator(messages) - tma_out = LLMAssistantResponseAggregator(messages) - - pipeline = Pipeline([ - transport.input(), # Transport user input - tma_in, # User responses - llm, # LLM - tts, # TTS - transport.output(), # Transport bot output - tma_out # Assistant spoken responses - ]) - - task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True, enable_metrics=True)) - - @transport.event_handler("on_first_participant_joined") - async def on_first_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) - # Kick off the conversation. - messages.append( - {"role": "system", "content": "Please introduce yourself to the user."}) - await task.queue_frames([LLMMessagesFrame(messages)]) - - runner = PipelineRunner() - - await runner.run(task) - - -if __name__ == "__main__": - asyncio.run(main()) diff --git a/pyproject.toml b/pyproject.toml index 73c643ddc..8a1e3a800 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -39,7 +39,7 @@ azure = [ "azure-cognitiveservices-speech~=1.40.0" ] cartesia = [ "websockets~=12.0" ] daily = [ "daily-python~=0.10.1" ] deepgram = [ "deepgram-sdk~=3.5.0" ] -elevenlabs = [ "elevenlabs~=1.7.0" ] +elevenlabs = [ "websockets~=12.0" ] examples = [ "python-dotenv~=1.0.1", "flask~=3.0.3", "flask_cors~=4.0.1" ] fal = [ "fal-client~=0.4.1" ] gladia = [ "websockets~=12.0" ] diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 72a8828c2..dcba578c5 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -6,7 +6,6 @@ import asyncio import io -import time import wave from abc import abstractmethod @@ -171,7 +170,7 @@ class TTSService(AIService): # if True, TTSService will push TTSStoppedFrames, otherwise subclass must do it push_stop_frames: bool = False, # if push_stop_frames is True, wait for this idle period before pushing TTSStoppedFrame - stop_frame_timeout_s: float = 0.8, + stop_frame_timeout_s: float = 1.0, **kwargs): super().__init__(**kwargs) self._aggregate_sentences: bool = aggregate_sentences @@ -319,16 +318,16 @@ class AsyncTTSService(TTSService): class AsyncWordTTSService(AsyncTTSService): def __init__(self, **kwargs): super().__init__(**kwargs) - self._start_word_timestamp = None + self._initial_word_timestamp = -1 self._words_queue = asyncio.Queue() self._words_task = self.get_event_loop().create_task(self._words_task_handler()) - def init_word_timestamps(self): - if not self._start_word_timestamp: - self._start_word_timestamp = self.get_clock().get_time() + def start_word_timestamps(self): + if self._initial_word_timestamp == -1: + self._initial_word_timestamp = self.get_clock().get_time() def reset_word_timestamps(self): - self._start_word_timestamp = None + self._initial_word_timestamp = -1 self._word_timestamps = [] async def add_word_timestamps(self, word_times: List[Tuple[str, float]]): @@ -366,7 +365,7 @@ class AsyncWordTTSService(AsyncTTSService): await self.push_frame(LLMFullResponseEndFrame()) else: frame = TextFrame(word) - frame.pts = self._start_word_timestamp + timestamp + frame.pts = self._initial_word_timestamp + timestamp await self.push_frame(frame) self._words_queue.task_done() except asyncio.CancelledError: diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index f263db60d..b15f266e9 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -181,7 +181,7 @@ class CartesiaTTSService(AsyncWordTTSService): ) elif msg["type"] == "chunk": await self.stop_ttfb_metrics() - self.init_word_timestamps() + self.start_word_timestamps() frame = AudioRawFrame( audio=base64.b64decode(msg["data"]), sample_rate=self._output_format["sample_rate"], diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index ed8041fcf..ded746144 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -4,17 +4,30 @@ # SPDX-License-Identifier: BSD 2-Clause License # -from typing import AsyncGenerator, Literal +import asyncio +import base64 +import json + +from typing import Any, AsyncGenerator, List, Literal, Mapping, Tuple from pydantic import BaseModel -from pipecat.frames.frames import AudioRawFrame, Frame, TTSStartedFrame, TTSStoppedFrame -from pipecat.services.ai_services import TTSService +from pipecat.frames.frames import ( + AudioRawFrame, + CancelFrame, + EndFrame, + Frame, + StartFrame, + StartInterruptionFrame, + TTSStartedFrame, + TTSStoppedFrame) +from pipecat.processors.frame_processor import FrameDirection +from pipecat.services.ai_services import AsyncWordTTSService from loguru import logger # See .env.example for ElevenLabs configuration needed try: - from elevenlabs.client import AsyncElevenLabs + import websockets except ModuleNotFoundError as e: logger.error(f"Exception: {e}") logger.error( @@ -35,7 +48,35 @@ def sample_rate_from_output_format(output_format: str) -> int: return 16000 -class ElevenLabsTTSService(TTSService): +def calculate_word_times( + alignment_info: Mapping[str, Any], cumulative_time: float +) -> List[Tuple[str, float]]: + end_times = [ + s + d for s, + d in zip( + alignment_info["charStartTimesMs"], + alignment_info["charDurationsMs"])] + zipped_end_times = list(zip(alignment_info["chars"], end_times)) + + # Get the start time of every character that appears after a space and + # match this to the word + words = "".join(alignment_info["chars"]).split(" ") + + # Calculate end time for each word. We do this by finding a space character + # and using the previous word time, also taking into account there might not + # be a space at the end. + times = [] + for (i, (a, b)) in enumerate(zipped_end_times): + if a == " " or i == len(zipped_end_times) - 1: + t = cumulative_time + (zipped_end_times[i - 1][1] / 1000.0) + times.append(t) + + word_times = list(zip(words, times)) + + return word_times + + +class ElevenLabsTTSService(AsyncWordTTSService): class InputParams(BaseModel): output_format: Literal["pcm_16000", "pcm_22050", "pcm_24000", "pcm_44100"] = "pcm_16000" @@ -45,49 +86,186 @@ class ElevenLabsTTSService(TTSService): api_key: str, voice_id: str, model: str = "eleven_turbo_v2_5", + url: str = "wss://api.elevenlabs.io", params: InputParams = InputParams(), **kwargs): - super().__init__(**kwargs) + # Aggregating sentences still gives cleaner-sounding results and fewer + # artifacts than streaming one word at a time. On average, waiting for a + # full sentence should only "cost" us 15ms or so with GPT-4o or a Llama + # 3 model, and it's worth it for the better audio quality. + # + # We also don't want to automatically push LLM response text frames, + # because the context aggregators will add them to the LLM context even + # if we're interrupted. ElevenLabs gives us word-by-word timestamps. We + # can use those to generate text frames ourselves aligned with the + # playout timing of the audio! + # + # Finally, ElevenLabs doesn't provide information on when the bot stops + # speaking for a while, so we want the parent class to send TTSStopFrame + # after a short period not receiving any audio. + super().__init__( + aggregate_sentences=True, + push_text_frames=False, + push_stop_frames=True, + stop_frame_timeout_s=2.0, + **kwargs + ) + self._api_key = api_key self._voice_id = voice_id self._model = model + self._url = url self._params = params - self._client = AsyncElevenLabs(api_key=api_key) self._sample_rate = sample_rate_from_output_format(params.output_format) + # Websocket connection to ElevenLabs. + self._websocket = None + # Indicates if we have sent TTSStartedFrame. It will reset to False when + # there's an interruption or TTSStoppedFrame. + self._started = False + self._cumulative_time = 0 + def can_generate_metrics(self) -> bool: return True async def set_model(self, model: str): logger.debug(f"Switching TTS model to: [{model}]") self._model = model + await self._disconnect() + await self._connect() async def set_voice(self, voice: str): logger.debug(f"Switching TTS voice to: [{voice}]") self._voice_id = voice + await self._disconnect() + await self._connect() + + async def start(self, frame: StartFrame): + await super().start(frame) + await self._connect() + + async def stop(self, frame: EndFrame): + await super().stop(frame) + await self._disconnect() + + async def cancel(self, frame: CancelFrame): + await super().cancel(frame) + await self._disconnect() + + async def flush_audio(self): + if self._websocket: + msg = {"text": " ", "flush": True} + await self._websocket.send(json.dumps(msg)) + + async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM): + await super().push_frame(frame, direction) + if isinstance(frame, (TTSStoppedFrame, StartInterruptionFrame)): + self._started = False + if isinstance(frame, TTSStoppedFrame): + await self.add_word_timestamps([("LLMFullResponseEndFrame", 0)]) + + async def _connect(self): + try: + voice_id = self._voice_id + model = self._model + output_format = self._params.output_format + url = f"{ + self._url}/v1/text-to-speech/{voice_id}/stream-input?model_id={model}&output_format={output_format}" + self._websocket = await websockets.connect(url) + self._receive_task = self.get_event_loop().create_task(self._receive_task_handler()) + self._keepalive_task = self.get_event_loop().create_task(self._keepalive_task_handler()) + + # According to ElevenLabs, we should always start with a single space. + msg = { + "text": " ", + "xi_api_key": self._api_key, + } + await self._websocket.send(json.dumps(msg)) + except Exception as e: + logger.exception(f"{self} initialization error: {e}") + self._websocket = None + + async def _disconnect(self): + try: + await self.stop_all_metrics() + + if self._receive_task: + self._receive_task.cancel() + await self._receive_task + self._receive_task = None + + if self._keepalive_task: + self._keepalive_task.cancel() + await self._keepalive_task + self._keepalive_task = None + + if self._websocket: + await self._websocket.close() + self._websocket = None + + self._started = False + except Exception as e: + logger.exception(f"{self} error closing websocket: {e}") + + async def _receive_task_handler(self): + 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 = AudioRawFrame(audio, self._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] + + except asyncio.CancelledError: + pass + except Exception as e: + logger.exception(f"{self} exception: {e}") + + async def _keepalive_task_handler(self): + while True: + try: + await asyncio.sleep(10) + await self._send_text("") + except asyncio.CancelledError: + break + except Exception as e: + logger.exception(f"{self} exception: {e}") + + async def _send_text(self, text: str): + if self._websocket: + msg = {"text": text + " "} + await self._websocket.send(json.dumps(msg)) async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: logger.debug(f"Generating TTS: [{text}]") - await self.start_tts_usage_metrics(text) - await self.start_ttfb_metrics() + try: + if not self._websocket: + await self._connect() - results = await self._client.generate( - text=text, - voice=self._voice_id, - model=self._model, - output_format=self._params.output_format - ) + try: + if not self._started: + await self.push_frame(TTSStartedFrame()) + await self.start_ttfb_metrics() + self._started = True + self._cumulative_time = 0 - tts_started = False - async for audio in results: - # This is so we send TTSStartedFrame when we have the first audio - # bytes. - if not tts_started: - await self.push_frame(TTSStartedFrame()) - tts_started = True - await self.stop_ttfb_metrics() - frame = AudioRawFrame(audio, self._sample_rate, 1) - yield frame - - await self.push_frame(TTSStoppedFrame()) + await self._send_text(text) + await self.start_tts_usage_metrics(text) + except Exception as e: + logger.error(f"{self} error sending message: {e}") + await self.push_frame(TTSStoppedFrame()) + await self._disconnect() + await self._connect() + return + yield None + except Exception as e: + logger.exception(f"{self} exception: {e}") diff --git a/src/pipecat/services/lmnt.py b/src/pipecat/services/lmnt.py index 59dd0aa5f..f5ad8aa1a 100644 --- a/src/pipecat/services/lmnt.py +++ b/src/pipecat/services/lmnt.py @@ -60,6 +60,8 @@ class LmntTTSService(AsyncTTSService): self._speech = None self._connection = None self._receive_task = None + # Indicates if we have sent TTSStartedFrame. It will reset to False when + # there's an interruption or TTSStoppedFrame. self._started = False def can_generate_metrics(self) -> bool: From 5acc4928febeb7e3566e67ab32b29bdae30ce73a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 13 Sep 2024 09:43:18 -0700 Subject: [PATCH 18/25] examples: add 07d-interruptible-elevenlabs.py --- .../07d-interruptible-elevenlabs.py | 99 +++++++++++++++++++ 1 file changed, 99 insertions(+) create mode 100644 examples/foundational/07d-interruptible-elevenlabs.py diff --git a/examples/foundational/07d-interruptible-elevenlabs.py b/examples/foundational/07d-interruptible-elevenlabs.py new file mode 100644 index 000000000..19bd4ad01 --- /dev/null +++ b/examples/foundational/07d-interruptible-elevenlabs.py @@ -0,0 +1,99 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import aiohttp +import os +import sys + +from pipecat.frames.frames import LLMMessagesFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.llm_response import ( + LLMAssistantResponseAggregator, LLMUserResponseAggregator) +from pipecat.services.elevenlabs import ElevenLabsTTSService +from pipecat.services.openai import OpenAILLMService +from pipecat.transports.services.daily import DailyParams, DailyTransport +from pipecat.vad.silero import SileroVADAnalyzer + +from runner import configure + +from loguru import logger + +from dotenv import load_dotenv +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + + +async def main(): + async with aiohttp.ClientSession() as session: + (room_url, token) = await configure(session) + + transport = DailyTransport( + room_url, + token, + "Respond bot", + DailyParams( + audio_out_enabled=True, + transcription_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer() + ) + ) + + tts = ElevenLabsTTSService( + api_key=os.getenv("ELEVENLABS_API_KEY", ""), + voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""), + ) + + llm = OpenAILLMService( + api_key=os.getenv("OPENAI_API_KEY"), + model="gpt-4o") + + 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.", + }, + ] + + tma_in = LLMUserResponseAggregator(messages) + tma_out = LLMAssistantResponseAggregator(messages) + + pipeline = Pipeline([ + transport.input(), # Transport user input + tma_in, # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + tma_out # Assistant spoken responses + ]) + + task = PipelineTask(pipeline, PipelineParams( + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + report_only_initial_ttfb=True, + )) + + @transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + transport.capture_participant_transcription(participant["id"]) + # Kick off the conversation. + messages.append( + {"role": "system", "content": "Please introduce yourself to the user."}) + await task.queue_frames([LLMMessagesFrame(messages)]) + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) From cb36a71381ea511ef84f17ac75d4177741761750 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 13 Sep 2024 09:50:05 -0700 Subject: [PATCH 19/25] fix some linting --- src/pipecat/frames/frames.py | 34 ++++------------------- src/pipecat/processors/frame_processor.py | 5 +--- src/pipecat/services/elevenlabs.py | 5 +--- 3 files changed, 7 insertions(+), 37 deletions(-) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 9cad1268d..51770dff1 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -55,13 +55,7 @@ class AudioRawFrame(DataFrame): def __str__(self): pts = format_pts(self.pts) - return f"{ - self.name}(pts: {pts}, size: { - len( - self.audio)}, frames: { - self.num_frames}, sample_rate: { - self.sample_rate}, channels: { - self.num_channels})" + return f"{self.name}(pts: {pts}, size: {len(self.audio)}, frames: {self.num_frames}, sample_rate: {self.sample_rate}, channels: {self.num_channels})" @dataclass @@ -102,11 +96,7 @@ class VisionImageRawFrame(ImageRawFrame): def __str__(self): pts = format_pts(self.pts) - return f"{ - self.name}(pts: {pts}, text: { - self.text}, size: { - self.size}, format: { - self.format})" + return f"{self.name}(pts: {pts}, text: {self.text}, size: {self.size}, format: {self.format})" @dataclass @@ -119,11 +109,7 @@ class UserImageRawFrame(ImageRawFrame): def __str__(self): pts = format_pts(self.pts) - return f"{ - self.name}(pts: {pts}, user: { - self.user_id}, size: { - self.size}, format: { - self.format})" + return f"{self.name}(pts: {pts}, user: {self.user_id}, size: {self.size}, format: {self.format})" @dataclass @@ -164,12 +150,7 @@ class TranscriptionFrame(TextFrame): language: Language | None = None def __str__(self): - return f"{ - self.name}(user: { - self.user_id}, text: { - self.text}, language: { - self.language}, timestamp: { - self.timestamp})" + return f"{self.name}(user: {self.user_id}, text: {self.text}, language: {self.language}, timestamp: {self.timestamp})" @dataclass @@ -181,12 +162,7 @@ class InterimTranscriptionFrame(TextFrame): language: Language | None = None def __str__(self): - return f"{ - self.name}(user: { - self.user_id}, text: { - self.text}, language: { - self.language}, timestamp: { - self.timestamp})" + return f"{self.name}(user: {self.user_id}, text: {self.text}, language: {self.language}, timestamp: {self.timestamp})" @dataclass diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index e228ec6b4..dfdee7d40 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -70,10 +70,7 @@ class FrameProcessorMetrics: async def start_llm_usage_metrics(self, tokens: dict): logger.debug( - f"{ - self._name} prompt tokens: { - tokens['prompt_tokens']}, completion tokens: { - tokens['completion_tokens']}") + f"{self._name} prompt tokens: {tokens['prompt_tokens']}, completion tokens: {tokens['completion_tokens']}") return MetricsFrame(tokens=[tokens]) async def start_tts_usage_metrics(self, text: str): diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index ded746144..ae3021ab6 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -58,8 +58,6 @@ def calculate_word_times( alignment_info["charDurationsMs"])] zipped_end_times = list(zip(alignment_info["chars"], end_times)) - # Get the start time of every character that appears after a space and - # match this to the word words = "".join(alignment_info["chars"]).split(" ") # Calculate end time for each word. We do this by finding a space character @@ -169,8 +167,7 @@ class ElevenLabsTTSService(AsyncWordTTSService): voice_id = self._voice_id model = self._model output_format = self._params.output_format - url = f"{ - self._url}/v1/text-to-speech/{voice_id}/stream-input?model_id={model}&output_format={output_format}" + url = f"{self._url}/v1/text-to-speech/{voice_id}/stream-input?model_id={model}&output_format={output_format}" self._websocket = await websockets.connect(url) self._receive_task = self.get_event_loop().create_task(self._receive_task_handler()) self._keepalive_task = self.get_event_loop().create_task(self._keepalive_task_handler()) From 1fe940bd6b6b2e8be70057bac390340341c9fc78 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 13 Sep 2024 13:10:44 -0700 Subject: [PATCH 20/25] servceis(cartesia,elevenlabs): use word start times instead --- src/pipecat/services/cartesia.py | 2 +- src/pipecat/services/elevenlabs.py | 15 +++++---------- 2 files changed, 6 insertions(+), 11 deletions(-) diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index b15f266e9..ac02ea469 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -177,7 +177,7 @@ class CartesiaTTSService(AsyncWordTTSService): await self.add_word_timestamps([("LLMFullResponseEndFrame", 0)]) elif msg["type"] == "timestamps": await self.add_word_timestamps( - list(zip(msg["word_timestamps"]["words"], msg["word_timestamps"]["end"])) + list(zip(msg["word_timestamps"]["words"], msg["word_timestamps"]["start"])) ) elif msg["type"] == "chunk": await self.stop_ttfb_metrics() diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index ae3021ab6..6a3b85914 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -51,22 +51,17 @@ def sample_rate_from_output_format(output_format: str) -> int: def calculate_word_times( alignment_info: Mapping[str, Any], cumulative_time: float ) -> List[Tuple[str, float]]: - end_times = [ - s + d for s, - d in zip( - alignment_info["charStartTimesMs"], - alignment_info["charDurationsMs"])] - zipped_end_times = list(zip(alignment_info["chars"], end_times)) + zipped_times = list(zip(alignment_info["chars"], alignment_info["charStartTimesMs"])) words = "".join(alignment_info["chars"]).split(" ") - # Calculate end time for each word. We do this by finding a space character + # Calculate start time for each word. We do this by finding a space character # and using the previous word time, also taking into account there might not # be a space at the end. times = [] - for (i, (a, b)) in enumerate(zipped_end_times): - if a == " " or i == len(zipped_end_times) - 1: - t = cumulative_time + (zipped_end_times[i - 1][1] / 1000.0) + for (i, (a, b)) in enumerate(zipped_times): + if a == " " or i == len(zipped_times) - 1: + t = cumulative_time + (zipped_times[i - 1][1] / 1000.0) times.append(t) word_times = list(zip(words, times)) From bce87f8717683393820cb806effc9a6fadd4337a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 13 Sep 2024 13:50:03 -0700 Subject: [PATCH 21/25] update CHANGELOG.md --- CHANGELOG.md | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9badb6c70..a757c7ffd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- A clock can now be specified to `PipelineTask` (defaults to + `SystemClock`). This clock will be passed to each frame processor via the + `StartFrame`. + +- Added pipeline clocks. A pipeline clock is used by the output transport to + know when a frame needs to be presented. For that, all frames now have an + optional `pts` field (prensentation timestamp). There's currently just one + clock implementation `SystemClock` and the `pts` field is currently only used + for `TextFrame`s (audio and image frames will be next). + - `DailyTransport` now supports setting the audio bitrate to improve audio quality through the `DailyParams.audio_out_bitrate` parameter. The new default is 96kbps. @@ -30,6 +40,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- `CartesiaTTSService` and `ElevenLabsTTSService` now add presentation + timestamps to their text output. This allows the output transport to push the + text frames downstream at almost the same time the words are spoken. We say + "almost" because currently the audio frames don't have presentation timestamp + but they should be played at roughly the same time. + - `DailyTransport.on_joined` event now returns the full session data instead of just the participant. From adaac003e544dec6107b244b6a2b2db4ebf8420e Mon Sep 17 00:00:00 2001 From: Kunal Shah Date: Mon, 16 Sep 2024 15:59:06 -0700 Subject: [PATCH 22/25] [Cartesia] Fix streaming truncation bug with Twilio Fast API WS --- src/pipecat/services/cartesia.py | 52 +++++++++++-------- .../transports/network/fastapi_websocket.py | 10 ++-- 2 files changed, 36 insertions(+), 26 deletions(-) diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index ac02ea469..6713a3825 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -4,33 +4,24 @@ # SPDX-License-Identifier: BSD 2-Clause License # -import json -import uuid -import base64 import asyncio +import base64 +import io +import json import time - +import uuid from typing import AsyncGenerator -from pipecat.frames.frames import ( - CancelFrame, - ErrorFrame, - Frame, - AudioRawFrame, - StartInterruptionFrame, - StartFrame, - EndFrame, - TTSStartedFrame, - TTSStoppedFrame, - TextFrame, - LLMFullResponseEndFrame -) -from pipecat.processors.frame_processor import FrameDirection -from pipecat.transcriptions.language import Language -from pipecat.services.ai_services import AsyncWordTTSService - from loguru import logger +from pipecat.frames.frames import (AudioRawFrame, CancelFrame, EndFrame, + ErrorFrame, Frame, LLMFullResponseEndFrame, + StartFrame, StartInterruptionFrame, + TextFrame, TTSStartedFrame, TTSStoppedFrame) +from pipecat.processors.frame_processor import FrameDirection +from pipecat.services.ai_services import AsyncWordTTSService +from pipecat.transcriptions.language import Language + # See .env.example for Cartesia configuration needed try: import websockets @@ -161,6 +152,25 @@ class CartesiaTTSService(AsyncWordTTSService): await self.push_frame(LLMFullResponseEndFrame()) self._context_id = None + async def flush_audio(self): + if not self._context_id or not self._websocket: + return + logger.debug("Flushing audio") + msg = { + "transcript": "", + "continue": False, + "context_id": self._context_id, + "model_id": self._model_id, + "voice": { + "mode": "id", + "id": self._voice_id + }, + "output_format": self._output_format, + "language": self._language, + "add_timestamps": True, + } + await self._websocket.send(json.dumps(msg)) + async def _receive_task_handler(self): try: async for message in self._websocket: diff --git a/src/pipecat/transports/network/fastapi_websocket.py b/src/pipecat/transports/network/fastapi_websocket.py index 16c5e81fc..b5bcfe7ed 100644 --- a/src/pipecat/transports/network/fastapi_websocket.py +++ b/src/pipecat/transports/network/fastapi_websocket.py @@ -8,19 +8,19 @@ import asyncio import io import wave - from typing import Awaitable, Callable + +from loguru import logger from pydantic.main import BaseModel -from pipecat.frames.frames import AudioRawFrame, CancelFrame, EndFrame, Frame, StartFrame, StartInterruptionFrame +from pipecat.frames.frames import (AudioRawFrame, CancelFrame, EndFrame, Frame, + StartFrame, StartInterruptionFrame) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.serializers.base_serializer import FrameSerializer from pipecat.transports.base_input import BaseInputTransport from pipecat.transports.base_output import BaseOutputTransport from pipecat.transports.base_transport import BaseTransport, TransportParams -from loguru import logger - try: from fastapi import WebSocket from starlette.websockets import WebSocketState @@ -101,7 +101,7 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport): async def write_raw_audio_frames(self, frames: bytes): self._websocket_audio_buffer += frames - while len(self._websocket_audio_buffer) >= self._params.audio_frame_size: + while len(self._websocket_audio_buffer): frame = AudioRawFrame( audio=self._websocket_audio_buffer[: self._params.audio_frame_size], From 0a26b650c000ae675eddd99b7871302613d87d75 Mon Sep 17 00:00:00 2001 From: Kunal Shah Date: Mon, 16 Sep 2024 16:06:25 -0700 Subject: [PATCH 23/25] Undo sorting --- src/pipecat/services/cartesia.py | 31 ++++++++++++++++++++----------- 1 file changed, 20 insertions(+), 11 deletions(-) diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index 6713a3825..a9d5aae67 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -4,23 +4,32 @@ # SPDX-License-Identifier: BSD 2-Clause License # -import asyncio -import base64 -import io import json -import time import uuid +import base64 +import asyncio +import time + from typing import AsyncGenerator -from loguru import logger - -from pipecat.frames.frames import (AudioRawFrame, CancelFrame, EndFrame, - ErrorFrame, Frame, LLMFullResponseEndFrame, - StartFrame, StartInterruptionFrame, - TextFrame, TTSStartedFrame, TTSStoppedFrame) +from pipecat.frames.frames import ( + CancelFrame, + ErrorFrame, + Frame, + AudioRawFrame, + StartInterruptionFrame, + StartFrame, + EndFrame, + TTSStartedFrame, + TTSStoppedFrame, + TextFrame, + LLMFullResponseEndFrame +) from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.ai_services import AsyncWordTTSService from pipecat.transcriptions.language import Language +from pipecat.services.ai_services import AsyncWordTTSService + +from loguru import logger # See .env.example for Cartesia configuration needed try: From 540cad4844711147699ee2f231b361125504a0e6 Mon Sep 17 00:00:00 2001 From: Kunal Shah Date: Mon, 16 Sep 2024 16:07:19 -0700 Subject: [PATCH 24/25] Undo sorting --- src/pipecat/transports/network/fastapi_websocket.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/pipecat/transports/network/fastapi_websocket.py b/src/pipecat/transports/network/fastapi_websocket.py index b5bcfe7ed..2c4bd187b 100644 --- a/src/pipecat/transports/network/fastapi_websocket.py +++ b/src/pipecat/transports/network/fastapi_websocket.py @@ -8,19 +8,19 @@ import asyncio import io import wave -from typing import Awaitable, Callable -from loguru import logger +from typing import Awaitable, Callable from pydantic.main import BaseModel -from pipecat.frames.frames import (AudioRawFrame, CancelFrame, EndFrame, Frame, - StartFrame, StartInterruptionFrame) +from pipecat.frames.frames import AudioRawFrame, CancelFrame, EndFrame, Frame, StartFrame, StartInterruptionFrame from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.serializers.base_serializer import FrameSerializer from pipecat.transports.base_input import BaseInputTransport from pipecat.transports.base_output import BaseOutputTransport from pipecat.transports.base_transport import BaseTransport, TransportParams +from loguru import logger + try: from fastapi import WebSocket from starlette.websockets import WebSocketState From 20c019ae1612a7aac76994a1842b3a80d407970b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 16 Sep 2024 23:54:21 -0700 Subject: [PATCH 25/25] services(cartesia,elevenlabs): close websocket before the receiving task --- src/pipecat/services/cartesia.py | 41 +++++++++++++++--------------- src/pipecat/services/elevenlabs.py | 20 +++++++-------- 2 files changed, 31 insertions(+), 30 deletions(-) diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index a9d5aae67..7b4463812 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -136,24 +136,25 @@ class CartesiaTTSService(AsyncWordTTSService): ) self._receive_task = self.get_event_loop().create_task(self._receive_task_handler()) except Exception as e: - logger.exception(f"{self} initialization error: {e}") + logger.error(f"{self} initialization error: {e}") self._websocket = None async def _disconnect(self): try: await self.stop_all_metrics() - if self._receive_task: - self._receive_task.cancel() - await self._receive_task - self._receive_task = None if self._websocket: await self._websocket.close() self._websocket = None + if self._receive_task: + self._receive_task.cancel() + await self._receive_task + self._receive_task = None + self._context_id = None except Exception as e: - logger.exception(f"{self} error closing websocket: {e}") + logger.error(f"{self} error closing websocket: {e}") async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection): await super()._handle_interruption(frame, direction) @@ -166,18 +167,18 @@ class CartesiaTTSService(AsyncWordTTSService): return logger.debug("Flushing audio") msg = { - "transcript": "", - "continue": False, - "context_id": self._context_id, - "model_id": self._model_id, - "voice": { - "mode": "id", - "id": self._voice_id - }, - "output_format": self._output_format, - "language": self._language, - "add_timestamps": True, - } + "transcript": "", + "continue": False, + "context_id": self._context_id, + "model_id": self._model_id, + "voice": { + "mode": "id", + "id": self._voice_id + }, + "output_format": self._output_format, + "language": self._language, + "add_timestamps": True, + } await self._websocket.send(json.dumps(msg)) async def _receive_task_handler(self): @@ -217,7 +218,7 @@ class CartesiaTTSService(AsyncWordTTSService): except asyncio.CancelledError: pass except Exception as e: - logger.exception(f"{self} exception: {e}") + logger.error(f"{self} exception: {e}") async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: logger.debug(f"Generating TTS: [{text}]") @@ -255,4 +256,4 @@ class CartesiaTTSService(AsyncWordTTSService): return yield None except Exception as e: - logger.exception(f"{self} exception: {e}") + logger.error(f"{self} exception: {e}") diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index 6a3b85914..a7a80033e 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -174,13 +174,18 @@ class ElevenLabsTTSService(AsyncWordTTSService): } await self._websocket.send(json.dumps(msg)) except Exception as e: - logger.exception(f"{self} initialization error: {e}") + logger.error(f"{self} initialization error: {e}") self._websocket = None async def _disconnect(self): try: await self.stop_all_metrics() + if self._websocket: + await self._websocket.send(json.dumps({"text": ""})) + await self._websocket.close() + self._websocket = None + if self._receive_task: self._receive_task.cancel() await self._receive_task @@ -191,13 +196,9 @@ class ElevenLabsTTSService(AsyncWordTTSService): await self._keepalive_task self._keepalive_task = None - if self._websocket: - await self._websocket.close() - self._websocket = None - self._started = False except Exception as e: - logger.exception(f"{self} error closing websocket: {e}") + logger.error(f"{self} error closing websocket: {e}") async def _receive_task_handler(self): try: @@ -215,11 +216,10 @@ class ElevenLabsTTSService(AsyncWordTTSService): word_times = calculate_word_times(msg["alignment"], self._cumulative_time) await self.add_word_timestamps(word_times) self._cumulative_time = word_times[-1][1] - except asyncio.CancelledError: pass except Exception as e: - logger.exception(f"{self} exception: {e}") + logger.error(f"{self} exception: {e}") async def _keepalive_task_handler(self): while True: @@ -229,7 +229,7 @@ class ElevenLabsTTSService(AsyncWordTTSService): except asyncio.CancelledError: break except Exception as e: - logger.exception(f"{self} exception: {e}") + logger.error(f"{self} exception: {e}") async def _send_text(self, text: str): if self._websocket: @@ -260,4 +260,4 @@ class ElevenLabsTTSService(AsyncWordTTSService): return yield None except Exception as e: - logger.exception(f"{self} exception: {e}") + logger.error(f"{self} exception: {e}")