From 1a542c91fadc8a168914132c01b2dff25fac070a Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Thu, 6 Jun 2024 10:48:22 -0400 Subject: [PATCH] temp commit, woring on playht --- .../07d-interruptible-cartesia.py | 4 + .../foundational/07e-interruptible-playht.py | 98 +++++++++++++++++++ src/pipecat/services/playht.py | 63 +++++++----- 3 files changed, 140 insertions(+), 25 deletions(-) create mode 100644 examples/foundational/07e-interruptible-playht.py diff --git a/examples/foundational/07d-interruptible-cartesia.py b/examples/foundational/07d-interruptible-cartesia.py index 39a77492b..a6398897a 100644 --- a/examples/foundational/07d-interruptible-cartesia.py +++ b/examples/foundational/07d-interruptible-cartesia.py @@ -19,6 +19,8 @@ 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 pipecat.processors.logger import FrameLogger + from runner import configure @@ -71,7 +73,9 @@ async def main(room_url: str, token): tma_in, # User responses llm, # LLM tts, # TTS + FrameLogger("tts out"), transport.output(), # Transport bot output + FrameLogger("transport out"), tma_out # Assistant spoken responses ]) diff --git a/examples/foundational/07e-interruptible-playht.py b/examples/foundational/07e-interruptible-playht.py new file mode 100644 index 000000000..5d3d63af2 --- /dev/null +++ b/examples/foundational/07e-interruptible-playht.py @@ -0,0 +1,98 @@ +# +# 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.playht import PlayHTTTSService +from pipecat.services.openai import OpenAILLMService +from pipecat.transports.services.daily import DailyParams, DailyTransport +from pipecat.vad.silero import SileroVADAnalyzer +from pipecat.processors.logger import FrameLogger + +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(room_url: str, token): + async with aiohttp.ClientSession() as session: + transport = DailyTransport( + room_url, + token, + "Respond bot", + DailyParams( + audio_out_enabled=True, + audio_out_sample_rate=16000, + transcription_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer() + ) + ) + + tts = PlayHTTTSService( + user_id=os.getenv("PLAYHT_USER_ID"), + api_key=os.getenv("PLAYHT_API_KEY"), + voice_url="s3://voice-cloning-zero-shot/801a663f-efd0-4254-98d0-5c175514c3e8/jennifer/manifest.json", + ) + + 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 + FrameLogger("tts out"), + transport.output(), # Transport bot output + FrameLogger("transport out"), + tma_out # Assistant spoken responses + ]) + + 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"]) + # 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__": + (url, token) = configure() + asyncio.run(main(url, token)) diff --git a/src/pipecat/services/playht.py b/src/pipecat/services/playht.py index e81cf1480..0c68512e4 100644 --- a/src/pipecat/services/playht.py +++ b/src/pipecat/services/playht.py @@ -6,6 +6,7 @@ import io import struct +import time from typing import AsyncGenerator @@ -25,7 +26,7 @@ except ModuleNotFoundError as e: raise Exception(f"Missing module: {e}") -class PlayHTAIService(TTSService): +class PlayHTTTSService(TTSService): def __init__(self, *, api_key: str, user_id: str, voice_url: str, **kwargs): super().__init__(**kwargs) @@ -47,28 +48,40 @@ class PlayHTAIService(TTSService): self._client.close() async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - b = bytearray() - in_header = True - for chunk in self._client.tts(text, self._options): - # skip the RIFF header. - if in_header: - b.extend(chunk) - if len(b) <= 36: - continue - else: - fh = io.BytesIO(b) - fh.seek(36) - (data, size) = struct.unpack('<4sI', fh.read(8)) - logger.debug( - f"first attempt: data: {data}, size: {hex(size)}, position: {fh.tell()}") - while data != b'data': - fh.read(size) + start_time = time.time() + ttfb = None + logger.debug(f"Generating TTS: [{text}]") + + try: + b = bytearray() + in_header = True + sync_gen = self._client.tts( + text, + voice_engine="PlayHT2.0-turbo", + options=self._options) + + # need to ask Aleix about this. frames are getting pushed. + # but playback is blocked + for chunk in sync_gen: + # skip the RIFF header. + if in_header: + b.extend(chunk) + if len(b) <= 36: + continue + else: + fh = io.BytesIO(b) + fh.seek(36) (data, size) = struct.unpack('<4sI', fh.read(8)) - logger.debug( - f"subsequent data: {data}, size: {hex(size)}, position: {fh.tell()}, data != data: {data != b'data'}") - logger.debug("position: ", fh.tell()) - in_header = False - else: - if len(chunk): - frame = AudioRawFrame(chunk, 16000, 1) - yield frame + while data != b'data': + fh.read(size) + (data, size) = struct.unpack('<4sI', fh.read(8)) + in_header = False + else: + if len(chunk): + if ttfb is None: + ttfb = time.time() - start_time + logger.debug(f"TTS ttfb: {ttfb}") + frame = AudioRawFrame(chunk, 16000, 1) + yield frame + except Exception as e: + logger.error(f"Error generating TTS: {e}")