From 8acf9a488bc15152be5104a711919323c1d93484 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 25 Feb 2025 12:30:54 -0800 Subject: [PATCH 1/3] tts: some small HTTP-based services improvements --- src/pipecat/services/elevenlabs.py | 8 +++++--- src/pipecat/services/openai.py | 4 +++- src/pipecat/services/playht.py | 13 ++++++------- src/pipecat/services/rime.py | 6 +++--- src/pipecat/services/xtts.py | 4 +++- 5 files changed, 20 insertions(+), 15 deletions(-) diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index 3dc42975e..5e70735f8 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -570,10 +570,12 @@ class ElevenLabsHttpTTSService(TTSService): await self.start_tts_usage_metrics(text) - yield TTSStartedFrame() + # Process the streaming response + CHUNK_SIZE = 1024 - async for chunk in response.content: - if chunk: + yield TTSStartedFrame() + async for chunk in response.content.iter_chunked(CHUNK_SIZE): + if len(chunk) > 0: await self.stop_ttfb_metrics() yield TTSAudioRawFrame(chunk, self.sample_rate, 1) except Exception as e: diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index 6ed3b4612..e49bd3a90 100644 --- a/src/pipecat/services/openai.py +++ b/src/pipecat/services/openai.py @@ -530,8 +530,10 @@ class OpenAITTSService(TTSService): await self.start_tts_usage_metrics(text) + CHUNK_SIZE = 1024 + yield TTSStartedFrame() - async for chunk in r.iter_bytes(8192): + async for chunk in r.iter_bytes(CHUNK_SIZE): if len(chunk) > 0: await self.stop_ttfb_metrics() frame = TTSAudioRawFrame(chunk, self.sample_rate, 1) diff --git a/src/pipecat/services/playht.py b/src/pipecat/services/playht.py index 56179e34c..c42a44eaa 100644 --- a/src/pipecat/services/playht.py +++ b/src/pipecat/services/playht.py @@ -383,8 +383,6 @@ class PlayHTHttpTTSService(TTSService): try: options = self._create_options() - b = bytearray() - in_header = True await self.start_ttfb_metrics() @@ -396,6 +394,8 @@ class PlayHTHttpTTSService(TTSService): yield TTSStartedFrame() + b = bytearray() + in_header = True async for chunk in playht_gen: # skip the RIFF header. if in_header: @@ -410,11 +410,10 @@ class PlayHTHttpTTSService(TTSService): fh.read(size) (data, size) = struct.unpack("<4sI", fh.read(8)) in_header = False - else: - if len(chunk): - await self.stop_ttfb_metrics() - frame = TTSAudioRawFrame(chunk, self.sample_rate, 1) - yield frame + elif len(chunk) > 0: + await self.stop_ttfb_metrics() + frame = TTSAudioRawFrame(chunk, self.sample_rate, 1) + yield frame except Exception as e: logger.error(f"{self} error generating TTS: {e}") finally: diff --git a/src/pipecat/services/rime.py b/src/pipecat/services/rime.py index 60d2d67ef..007ba958e 100644 --- a/src/pipecat/services/rime.py +++ b/src/pipecat/services/rime.py @@ -407,10 +407,10 @@ class RimeHttpTTSService(TTSService): yield TTSStartedFrame() # Process the streaming response - chunk_size = 8192 + CHUNK_SIZE = 1024 - async for chunk in response.content.iter_chunked(chunk_size): - if chunk: + async for chunk in response.content.iter_chunked(CHUNK_SIZE): + if len(chunk) > 0: await self.stop_ttfb_metrics() frame = TTSAudioRawFrame(chunk, self.sample_rate, 1) yield frame diff --git a/src/pipecat/services/xtts.py b/src/pipecat/services/xtts.py index 9d2f3dd1e..e275b50b0 100644 --- a/src/pipecat/services/xtts.py +++ b/src/pipecat/services/xtts.py @@ -150,8 +150,10 @@ class XTTSService(TTSService): yield TTSStartedFrame() + CHUNK_SIZE = 1024 + buffer = bytearray() - async for chunk in r.content.iter_chunked(1024): + async for chunk in r.content.iter_chunked(CHUNK_SIZE): if len(chunk) > 0: await self.stop_ttfb_metrics() # Append new chunk to the buffer. From 27161f8e3b4e8db40660cb2c36f5663a1f71b02c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 25 Feb 2025 13:06:43 -0800 Subject: [PATCH 2/3] BaseOutputTransport: cleanup audio buffer after bot stops talking --- src/pipecat/transports/base_output.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 2322ade9d..a2ee5aa92 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -232,6 +232,9 @@ class BaseOutputTransport(FrameProcessor): await self.push_frame(BotStoppedSpeakingFrame()) await self.push_frame(BotStoppedSpeakingFrame(), FrameDirection.UPSTREAM) self._bot_speaking = False + # Clean audio buffer (there could be tiny left overs if not multiple + # to our output chunk size). + self._audio_buffer = bytearray() # # Sink tasks From b994a03466d8d15177c8ae9f95542f6bb49d5b9a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 25 Feb 2025 21:40:41 -0800 Subject: [PATCH 3/3] examples: add more HTTP TTS services examples --- .../07d-interruptible-elevenlabs-http.py | 103 ++++++++++++++++++ .../07q-interruptible-rime-http.py | 103 ++++++++++++++++++ 2 files changed, 206 insertions(+) create mode 100644 examples/foundational/07d-interruptible-elevenlabs-http.py create mode 100644 examples/foundational/07q-interruptible-rime-http.py diff --git a/examples/foundational/07d-interruptible-elevenlabs-http.py b/examples/foundational/07d-interruptible-elevenlabs-http.py new file mode 100644 index 000000000..63894851f --- /dev/null +++ b/examples/foundational/07d-interruptible-elevenlabs-http.py @@ -0,0 +1,103 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import os +import sys + +import aiohttp +from dotenv import load_dotenv +from loguru import logger +from runner import configure + +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.services.elevenlabs import ElevenLabsHttpTTSService +from pipecat.services.openai import OpenAILLMService +from pipecat.transports.services.daily import DailyParams, DailyTransport + +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 = ElevenLabsHttpTTSService( + api_key=os.getenv("ELEVENLABS_API_KEY", ""), + voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""), + aiohttp_session=session, + ) + + 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.", + }, + ] + + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + context_aggregator.user(), # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses + ] + ) + + task = PipelineTask( + pipeline, + params=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): + await 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([context_aggregator.user().get_context_frame()]) + + @transport.event_handler("on_participant_left") + async def on_participant_left(transport, participant, reason): + await task.cancel() + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/foundational/07q-interruptible-rime-http.py b/examples/foundational/07q-interruptible-rime-http.py new file mode 100644 index 000000000..eaa49179e --- /dev/null +++ b/examples/foundational/07q-interruptible-rime-http.py @@ -0,0 +1,103 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import os +import sys + +import aiohttp +from dotenv import load_dotenv +from loguru import logger +from runner import configure + +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.services.openai import OpenAILLMService +from pipecat.services.rime import RimeHttpTTSService +from pipecat.transports.services.daily import DailyParams, DailyTransport + +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 = RimeHttpTTSService( + api_key=os.getenv("RIME_API_KEY", ""), + voice_id="rex", + aiohttp_session=session, + ) + + 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.", + }, + ] + + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + context_aggregator.user(), # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses + ] + ) + + task = PipelineTask( + pipeline, + params=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): + await 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([context_aggregator.user().get_context_frame()]) + + @transport.event_handler("on_participant_left") + async def on_participant_left(transport, participant, reason): + await task.cancel() + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main())