From 06ff9cfede9c9cbc1373a97afa96eb5d9b3a05ca Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Wed, 5 Jun 2024 13:33:02 -0400 Subject: [PATCH 01/10] added timing logs for cartesia, deepgram, elevenlabs --- src/pipecat/services/cartesia.py | 6 ++++++ src/pipecat/services/deepgram.py | 6 ++++++ src/pipecat/services/elevenlabs.py | 6 ++++++ 3 files changed, 18 insertions(+) diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index f2d8c9b14..9c730a283 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -3,6 +3,7 @@ # # SPDX-License-Identifier: BSD 2-Clause License # +import time from cartesia.tts import AsyncCartesiaTTS @@ -40,6 +41,8 @@ class CartesiaTTSService(TTSService): logger.error(f"Cartesia initialization error: {e}") async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + start_time = time.time() + ttfb = None logger.debug(f"Generating TTS: [{text}]") try: @@ -52,6 +55,9 @@ class CartesiaTTSService(TTSService): ) async for chunk in chunk_generator: + if ttfb is None: + ttfb = time.time() - start_time + logger.debug(f"TTS ttfb: {ttfb}") yield AudioRawFrame(chunk["audio"], chunk["sampling_rate"], 1) except Exception as e: logger.error(f"Cartesia exception: {e}") diff --git a/src/pipecat/services/deepgram.py b/src/pipecat/services/deepgram.py index b5901825e..7b19e04e2 100644 --- a/src/pipecat/services/deepgram.py +++ b/src/pipecat/services/deepgram.py @@ -5,6 +5,7 @@ # import aiohttp +import time from typing import AsyncGenerator @@ -30,6 +31,8 @@ class DeepgramTTSService(TTSService): self._aiohttp_session = aiohttp_session async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + start_time = time.time() + ttfb = None logger.debug(f"Generating TTS: [{text}]") base_url = "https://api.deepgram.com/v1/speak" @@ -46,6 +49,9 @@ class DeepgramTTSService(TTSService): return async for data in r.content: + if ttfb is None: + ttfb = time.time() - start_time + logger.debug(f"TTS ttfb: {ttfb}") frame = AudioRawFrame(audio=data, sample_rate=16000, num_channels=1) yield frame except Exception as e: diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index 3d602595d..d5b476160 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -5,6 +5,7 @@ # import aiohttp +import time from typing import AsyncGenerator @@ -32,6 +33,8 @@ class ElevenLabsTTSService(TTSService): self._model = model async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + start_time = time.time() + ttfb = None logger.debug(f"Generating TTS: [{text}]") url = f"https://api.elevenlabs.io/v1/text-to-speech/{self._voice_id}/stream" @@ -56,5 +59,8 @@ class ElevenLabsTTSService(TTSService): async for chunk in r.content: if len(chunk) > 0: + if ttfb is None: + ttfb = time.time() - start_time + logger.debug(f"TTS ttfb: {ttfb}") frame = AudioRawFrame(chunk, 16000, 1) yield frame From 1a542c91fadc8a168914132c01b2dff25fac070a Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Thu, 6 Jun 2024 10:48:22 -0400 Subject: [PATCH 02/10] 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}") From 3eff1e559b81ee394dad6b3cec1d9a475ae40586 Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Thu, 6 Jun 2024 11:11:06 -0400 Subject: [PATCH 03/10] pipecat async working, but maybe needs a threaded implementation --- src/pipecat/services/playht.py | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/src/pipecat/services/playht.py b/src/pipecat/services/playht.py index 0c68512e4..b0f0ce5de 100644 --- a/src/pipecat/services/playht.py +++ b/src/pipecat/services/playht.py @@ -7,6 +7,7 @@ import io import struct import time +import asyncio from typing import AsyncGenerator @@ -52,6 +53,11 @@ class PlayHTTTSService(TTSService): ttfb = None logger.debug(f"Generating TTS: [{text}]") + async def async_generator(sync_gen): + for item in sync_gen: + await asyncio.sleep(0) # Yield control back to the event loop + yield item + try: b = bytearray() in_header = True @@ -62,7 +68,9 @@ class PlayHTTTSService(TTSService): # need to ask Aleix about this. frames are getting pushed. # but playback is blocked - for chunk in sync_gen: + + # for chunk in sync_gen: + async for chunk in async_generator(sync_gen): # skip the RIFF header. if in_header: b.extend(chunk) From aee3916cd1812b0ebfdc6cfc0909c21d61dc1b53 Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Thu, 6 Jun 2024 11:24:26 -0400 Subject: [PATCH 04/10] cartesia async fixed --- .../foundational/07d-interruptible-cartesia.py | 3 --- examples/foundational/07e-interruptible-playht.py | 2 -- src/pipecat/services/playht.py | 14 ++++---------- 3 files changed, 4 insertions(+), 15 deletions(-) diff --git a/examples/foundational/07d-interruptible-cartesia.py b/examples/foundational/07d-interruptible-cartesia.py index a6398897a..283baa49a 100644 --- a/examples/foundational/07d-interruptible-cartesia.py +++ b/examples/foundational/07d-interruptible-cartesia.py @@ -19,7 +19,6 @@ 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 @@ -73,9 +72,7 @@ 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 index 5d3d63af2..c6c062e52 100644 --- a/examples/foundational/07e-interruptible-playht.py +++ b/examples/foundational/07e-interruptible-playht.py @@ -72,9 +72,7 @@ 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/src/pipecat/services/playht.py b/src/pipecat/services/playht.py index b0f0ce5de..8373eca7b 100644 --- a/src/pipecat/services/playht.py +++ b/src/pipecat/services/playht.py @@ -17,8 +17,8 @@ from pipecat.services.ai_services import TTSService from loguru import logger try: - from pyht import Client from pyht.client import TTSOptions + from pyht.async_client import AsyncClient from pyht.protos.api_pb2 import Format except ModuleNotFoundError as e: logger.error(f"Exception: {e}") @@ -35,7 +35,7 @@ class PlayHTTTSService(TTSService): self._user_id = user_id self._speech_key = api_key - self._client = Client( + self._client = AsyncClient( user_id=self._user_id, api_key=self._speech_key, ) @@ -53,15 +53,10 @@ class PlayHTTTSService(TTSService): ttfb = None logger.debug(f"Generating TTS: [{text}]") - async def async_generator(sync_gen): - for item in sync_gen: - await asyncio.sleep(0) # Yield control back to the event loop - yield item - try: b = bytearray() in_header = True - sync_gen = self._client.tts( + playht_gen = self._client.tts( text, voice_engine="PlayHT2.0-turbo", options=self._options) @@ -69,8 +64,7 @@ class PlayHTTTSService(TTSService): # need to ask Aleix about this. frames are getting pushed. # but playback is blocked - # for chunk in sync_gen: - async for chunk in async_generator(sync_gen): + async for chunk in playht_gen: # skip the RIFF header. if in_header: b.extend(chunk) From ddfd721f6ec2bb189a71925399395e52ffc45a16 Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Thu, 6 Jun 2024 11:32:47 -0400 Subject: [PATCH 05/10] openai tts ttfb --- .../07g-interruptible-openai-tts.py | 94 +++++++++++++++++++ src/pipecat/services/openai.py | 5 + 2 files changed, 99 insertions(+) create mode 100644 examples/foundational/07g-interruptible-openai-tts.py diff --git a/examples/foundational/07g-interruptible-openai-tts.py b/examples/foundational/07g-interruptible-openai-tts.py new file mode 100644 index 000000000..2a45b63d8 --- /dev/null +++ b/examples/foundational/07g-interruptible-openai-tts.py @@ -0,0 +1,94 @@ +# +# 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.openai import OpenAITTSService +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(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=24000, + transcription_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer() + ) + ) + + tts = OpenAITTSService( + api_key=os.getenv("OPENAI_API_KEY"), + voice="alloy" + ) + + 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)) + + @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/openai.py b/src/pipecat/services/openai.py index 2937eb61d..4dca1a998 100644 --- a/src/pipecat/services/openai.py +++ b/src/pipecat/services/openai.py @@ -304,6 +304,8 @@ class OpenAITTSService(TTSService): self._client = AsyncOpenAI(api_key=api_key) async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + start_time = time.time() + ttfb = None logger.debug(f"Generating TTS: [{text}]") try: @@ -320,6 +322,9 @@ class OpenAITTSService(TTSService): return async for chunk in r.iter_bytes(8192): if len(chunk) > 0: + if ttfb is None: + ttfb = time.time() - start_time + logger.debug(f"TTS ttfb: {ttfb}") frame = AudioRawFrame(chunk, 24_000, 1) yield frame except BadRequestError as e: From ac7bc359444bd65b3204404356354d137094772f Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Thu, 6 Jun 2024 11:45:48 -0400 Subject: [PATCH 06/10] azure tts ttfb --- .../07f-interruptible-azure-tts.py | 95 +++++++++++++++++++ src/pipecat/services/azure.py | 6 ++ 2 files changed, 101 insertions(+) create mode 100644 examples/foundational/07f-interruptible-azure-tts.py diff --git a/examples/foundational/07f-interruptible-azure-tts.py b/examples/foundational/07f-interruptible-azure-tts.py new file mode 100644 index 000000000..1770a3213 --- /dev/null +++ b/examples/foundational/07f-interruptible-azure-tts.py @@ -0,0 +1,95 @@ +# +# 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.azure import AzureTTSService +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(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 = AzureTTSService( + api_key=os.getenv("AZURE_SPEECH_API_KEY"), + region=os.getenv("AZURE_SPEECH_REGION"), + ) + + 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)) + + @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/azure.py b/src/pipecat/services/azure.py index 216d0555b..fce3c5939 100644 --- a/src/pipecat/services/azure.py +++ b/src/pipecat/services/azure.py @@ -7,6 +7,7 @@ import aiohttp import asyncio import io +import time from PIL import Image from typing import AsyncGenerator @@ -46,6 +47,8 @@ class AzureTTSService(TTSService): self._voice = voice async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + start_time = time.time() + ttfb = None logger.debug(f"Generating TTS: {text}") ssml = ( @@ -61,6 +64,9 @@ class AzureTTSService(TTSService): result = await asyncio.to_thread(self.speech_synthesizer.speak_ssml, (ssml)) if result.reason == ResultReason.SynthesizingAudioCompleted: + if ttfb is None: + ttfb = time.time() - start_time + logger.debug(f"TTS ttfb: {ttfb}") # Azure always sends a 44-byte header. Strip it off. yield AudioRawFrame(audio=result.audio_data[44:], sample_rate=16000, num_channels=1) elif result.reason == ResultReason.Canceled: From a5eb30a93dcd052ddbf449340115730391c54352 Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Thu, 6 Jun 2024 11:49:05 -0400 Subject: [PATCH 07/10] changelog --- CHANGELOG.md | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index d46b012c3..1f9f6d10d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,16 @@ All notable changes to **pipecat** will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## Unreleased + +### Added + +- Added TTFB debug logging for TTS services + +### Fixed + +- Fixed PlayHT TTS service to work properly async + ## [0.0.27] - 2024-06-05 ### Added From e765a29ca2f65c0e4b938a043fc8add24c9ccbb4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 6 Jun 2024 10:53:32 -0700 Subject: [PATCH 08/10] processors: implement base process_frame(). all subsclassed should call it --- .../foundational/05-sync-speech-and-image.py | 2 ++ .../05a-local-sync-speech-and-image.py | 6 ++++++ examples/foundational/06a-image-sync.py | 2 ++ examples/foundational/11-sound-effects.py | 4 ++++ examples/foundational/12-describe-video.py | 2 ++ .../12a-describe-video-gemini-flash.py | 2 ++ .../foundational/12b-describe-video-gpt-4o.py | 2 ++ .../12c-describe-video-anthropic.py | 2 ++ .../foundational/13-whisper-transcription.py | 2 ++ examples/foundational/13a-whisper-local.py | 2 ++ examples/moondream-chatbot/bot.py | 8 ++++++++ examples/simple-chatbot/bot.py | 2 ++ .../storytelling-chatbot/src/processors.py | 4 ++++ examples/translation-chatbot/bot.py | 4 ++++ src/pipecat/frames/frames.py | 3 ++- src/pipecat/pipeline/parallel_pipeline.py | 6 ++++++ src/pipecat/pipeline/pipeline.py | 6 ++++++ src/pipecat/pipeline/task.py | 2 ++ src/pipecat/processors/aggregators/gated.py | 2 ++ .../processors/aggregators/llm_response.py | 4 ++++ .../processors/aggregators/parallel_task.py | 6 ++++++ src/pipecat/processors/aggregators/sentence.py | 2 ++ .../processors/aggregators/user_response.py | 2 ++ .../aggregators/vision_image_frame.py | 2 ++ src/pipecat/processors/filters/frame_filter.py | 2 ++ .../processors/filters/wake_check_filter.py | 2 ++ src/pipecat/processors/frame_processor.py | 18 ++++++++++++++++-- src/pipecat/processors/frameworks/langchain.py | 2 ++ src/pipecat/processors/text_transformer.py | 2 ++ src/pipecat/services/ai_services.py | 8 ++++++++ src/pipecat/services/anthropic.py | 2 ++ src/pipecat/services/google.py | 2 ++ src/pipecat/services/openai.py | 2 ++ src/pipecat/transports/base_input.py | 11 +++-------- src/pipecat/transports/base_output.py | 10 +++------- src/pipecat/transports/services/daily.py | 2 ++ src/pipecat/utils/test_frame_processor.py | 2 ++ src/pipecat/vad/silero.py | 2 ++ tests/test_langchain.py | 2 ++ 39 files changed, 130 insertions(+), 18 deletions(-) diff --git a/examples/foundational/05-sync-speech-and-image.py b/examples/foundational/05-sync-speech-and-image.py index f057c847c..1c2e97e8b 100644 --- a/examples/foundational/05-sync-speech-and-image.py +++ b/examples/foundational/05-sync-speech-and-image.py @@ -59,6 +59,8 @@ class MonthPrepender(FrameProcessor): self.prepend_to_next_text_frame = False async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, MonthFrame): self.most_recent_month = frame.month elif self.prepend_to_next_text_frame and isinstance(frame, TextFrame): diff --git a/examples/foundational/05a-local-sync-speech-and-image.py b/examples/foundational/05a-local-sync-speech-and-image.py index a377ebe98..7db6dc99f 100644 --- a/examples/foundational/05a-local-sync-speech-and-image.py +++ b/examples/foundational/05a-local-sync-speech-and-image.py @@ -50,6 +50,8 @@ async def main(): self.text = "" async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, TextFrame): self.text = frame.text await self.push_frame(frame, direction) @@ -60,6 +62,8 @@ async def main(): self.audio = bytearray() async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, AudioRawFrame): self.audio.extend(frame.audio) self.frame = AudioRawFrame( @@ -71,6 +75,8 @@ async def main(): self.frame = None async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, URLImageRawFrame): self.frame = frame diff --git a/examples/foundational/06a-image-sync.py b/examples/foundational/06a-image-sync.py index 2f5528ee4..686964502 100644 --- a/examples/foundational/06a-image-sync.py +++ b/examples/foundational/06a-image-sync.py @@ -49,6 +49,8 @@ class ImageSyncAggregator(FrameProcessor): self._waiting_image_bytes = self._waiting_image.tobytes() async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if not isinstance(frame, SystemFrame): await self.push_frame(ImageRawFrame(image=self._speaking_image_bytes, size=(1024, 1024), format=self._speaking_image_format)) await self.push_frame(frame) diff --git a/examples/foundational/11-sound-effects.py b/examples/foundational/11-sound-effects.py index 2a3e8effc..8abf3d935 100644 --- a/examples/foundational/11-sound-effects.py +++ b/examples/foundational/11-sound-effects.py @@ -60,6 +60,8 @@ for file in sound_files: class OutboundSoundEffectWrapper(FrameProcessor): async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, LLMFullResponseEndFrame): await self.push_frame(sounds["ding1.wav"]) # In case anything else downstream needs it @@ -71,6 +73,8 @@ class OutboundSoundEffectWrapper(FrameProcessor): class InboundSoundEffectWrapper(FrameProcessor): async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, LLMMessagesFrame): await self.push_frame(sounds["ding2.wav"]) # In case anything else downstream needs it diff --git a/examples/foundational/12-describe-video.py b/examples/foundational/12-describe-video.py index 256580c07..bde70d664 100644 --- a/examples/foundational/12-describe-video.py +++ b/examples/foundational/12-describe-video.py @@ -42,6 +42,8 @@ class UserImageRequester(FrameProcessor): self._participant_id = participant_id async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if self._participant_id and isinstance(frame, TextFrame): await self.push_frame(UserImageRequestFrame(self._participant_id), FrameDirection.UPSTREAM) await self.push_frame(frame, direction) diff --git a/examples/foundational/12a-describe-video-gemini-flash.py b/examples/foundational/12a-describe-video-gemini-flash.py index 00e168ad5..293e9a26a 100644 --- a/examples/foundational/12a-describe-video-gemini-flash.py +++ b/examples/foundational/12a-describe-video-gemini-flash.py @@ -42,6 +42,8 @@ class UserImageRequester(FrameProcessor): self._participant_id = participant_id async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if self._participant_id and isinstance(frame, TextFrame): await self.push_frame(UserImageRequestFrame(self._participant_id), FrameDirection.UPSTREAM) await self.push_frame(frame, direction) diff --git a/examples/foundational/12b-describe-video-gpt-4o.py b/examples/foundational/12b-describe-video-gpt-4o.py index c34a5a94a..19caa307c 100644 --- a/examples/foundational/12b-describe-video-gpt-4o.py +++ b/examples/foundational/12b-describe-video-gpt-4o.py @@ -42,6 +42,8 @@ class UserImageRequester(FrameProcessor): self._participant_id = participant_id async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if self._participant_id and isinstance(frame, TextFrame): await self.push_frame(UserImageRequestFrame(self._participant_id), FrameDirection.UPSTREAM) await self.push_frame(frame, direction) diff --git a/examples/foundational/12c-describe-video-anthropic.py b/examples/foundational/12c-describe-video-anthropic.py index 9a7a9a804..033af2a0a 100644 --- a/examples/foundational/12c-describe-video-anthropic.py +++ b/examples/foundational/12c-describe-video-anthropic.py @@ -42,6 +42,8 @@ class UserImageRequester(FrameProcessor): self._participant_id = participant_id async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if self._participant_id and isinstance(frame, TextFrame): await self.push_frame(UserImageRequestFrame(self._participant_id), FrameDirection.UPSTREAM) await self.push_frame(frame, direction) diff --git a/examples/foundational/13-whisper-transcription.py b/examples/foundational/13-whisper-transcription.py index 2ca8eb9fa..43d5f4840 100644 --- a/examples/foundational/13-whisper-transcription.py +++ b/examples/foundational/13-whisper-transcription.py @@ -29,6 +29,8 @@ logger.add(sys.stderr, level="DEBUG") class TranscriptionLogger(FrameProcessor): async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, TranscriptionFrame): print(f"Transcription: {frame.text}") diff --git a/examples/foundational/13a-whisper-local.py b/examples/foundational/13a-whisper-local.py index 366cd2cf5..6bf27aa0a 100644 --- a/examples/foundational/13a-whisper-local.py +++ b/examples/foundational/13a-whisper-local.py @@ -28,6 +28,8 @@ logger.add(sys.stderr, level="DEBUG") class TranscriptionLogger(FrameProcessor): async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, TranscriptionFrame): print(f"Transcription: {frame.text}") diff --git a/examples/moondream-chatbot/bot.py b/examples/moondream-chatbot/bot.py index 3e9ced259..8407d04cc 100644 --- a/examples/moondream-chatbot/bot.py +++ b/examples/moondream-chatbot/bot.py @@ -74,6 +74,8 @@ class TalkingAnimation(FrameProcessor): self._is_talking = False async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, AudioRawFrame): if not self._is_talking: await self.push_frame(talking_frame) @@ -93,6 +95,8 @@ class UserImageRequester(FrameProcessor): self.participant_id = participant_id async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if self.participant_id and isinstance(frame, TextFrame): if frame.text == user_request_answer: await self.push_frame(UserImageRequestFrame(self.participant_id), FrameDirection.UPSTREAM) @@ -107,6 +111,8 @@ class TextFilterProcessor(FrameProcessor): self.text = text async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, TextFrame): if frame.text != self.text: await self.push_frame(frame) @@ -116,6 +122,8 @@ class TextFilterProcessor(FrameProcessor): class ImageFilterProcessor(FrameProcessor): async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if not isinstance(frame, ImageRawFrame): await self.push_frame(frame) diff --git a/examples/simple-chatbot/bot.py b/examples/simple-chatbot/bot.py index 80b60833f..778902a18 100644 --- a/examples/simple-chatbot/bot.py +++ b/examples/simple-chatbot/bot.py @@ -64,6 +64,8 @@ class TalkingAnimation(FrameProcessor): self._is_talking = False async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, AudioRawFrame): if not self._is_talking: await self.push_frame(talking_frame) diff --git a/examples/storytelling-chatbot/src/processors.py b/examples/storytelling-chatbot/src/processors.py index 18428eb72..a8b2a0980 100644 --- a/examples/storytelling-chatbot/src/processors.py +++ b/examples/storytelling-chatbot/src/processors.py @@ -52,6 +52,8 @@ class StoryImageProcessor(FrameProcessor): self._fal_service = fal_service async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, StoryImageFrame): try: async with timeout(7): @@ -86,6 +88,8 @@ class StoryProcessor(FrameProcessor): self._story = story async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, UserStoppedSpeakingFrame): # Send an app message to the UI await self.push_frame(DailyTransportMessageFrame(CUE_ASSISTANT_TURN)) diff --git a/examples/translation-chatbot/bot.py b/examples/translation-chatbot/bot.py index 38667d897..9354dc0be 100644 --- a/examples/translation-chatbot/bot.py +++ b/examples/translation-chatbot/bot.py @@ -40,6 +40,8 @@ class TranslationProcessor(FrameProcessor): self._language = language async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, TextFrame): context = [ { @@ -65,6 +67,8 @@ class TranslationSubtitles(FrameProcessor): # subtitles. # async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, TextFrame): message = { "language": self._language, diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 8eb32664c..9a5725de1 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -4,7 +4,7 @@ # SPDX-License-Identifier: BSD 2-Clause License # -from typing import Any, List, Tuple +from typing import Any, List, Mapping, Tuple from dataclasses import dataclass, field @@ -188,6 +188,7 @@ class SystemFrame(Frame): class StartFrame(SystemFrame): """This is the first frame that should be pushed down a pipeline.""" allow_interruptions: bool = False + enable_metrics: bool = False @dataclass diff --git a/src/pipecat/pipeline/parallel_pipeline.py b/src/pipecat/pipeline/parallel_pipeline.py index 22ffdfdf2..ccf72bd90 100644 --- a/src/pipecat/pipeline/parallel_pipeline.py +++ b/src/pipecat/pipeline/parallel_pipeline.py @@ -20,6 +20,8 @@ class Source(FrameProcessor): self._up_queue = upstream_queue async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + match direction: case FrameDirection.UPSTREAM: await self._up_queue.put(frame) @@ -34,6 +36,8 @@ class Sink(FrameProcessor): self._down_queue = downstream_queue async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + match direction: case FrameDirection.UPSTREAM: await self.push_frame(frame, direction) @@ -90,6 +94,8 @@ class ParallelPipeline(FrameProcessor): self._down_task = loop.create_task(self._process_down_queue()) async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, StartFrame): await self._start_tasks() diff --git a/src/pipecat/pipeline/pipeline.py b/src/pipecat/pipeline/pipeline.py index 2ba061a37..2cb5b45d4 100644 --- a/src/pipecat/pipeline/pipeline.py +++ b/src/pipecat/pipeline/pipeline.py @@ -19,6 +19,8 @@ class PipelineSource(FrameProcessor): self._upstream_push_frame = upstream_push_frame async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + match direction: case FrameDirection.UPSTREAM: await self._upstream_push_frame(frame, direction) @@ -33,6 +35,8 @@ class PipelineSink(FrameProcessor): self._downstream_push_frame = downstream_push_frame async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + match direction: case FrameDirection.UPSTREAM: await self.push_frame(frame, direction) @@ -61,6 +65,8 @@ class Pipeline(FrameProcessor): await self._cleanup_processors() async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if direction == FrameDirection.DOWNSTREAM: await self._source.process_frame(frame, FrameDirection.DOWNSTREAM) elif direction == FrameDirection.UPSTREAM: diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index e3e7f7e36..b1f72c096 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -28,6 +28,8 @@ class Source(FrameProcessor): self._up_queue = up_queue async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + match direction: case FrameDirection.UPSTREAM: await self._up_queue.put(frame) diff --git a/src/pipecat/processors/aggregators/gated.py b/src/pipecat/processors/aggregators/gated.py index 3c80e4641..16a7e3076 100644 --- a/src/pipecat/processors/aggregators/gated.py +++ b/src/pipecat/processors/aggregators/gated.py @@ -48,6 +48,8 @@ class GatedAggregator(FrameProcessor): self._accumulator: List[Frame] = [] async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + # We must not block system frames. if isinstance(frame, SystemFrame): await self.push_frame(frame, direction) diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 289296487..fdab187bd 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -79,6 +79,8 @@ class LLMResponseAggregator(FrameProcessor): # and T2 would be dropped. async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + send_aggregation = False if isinstance(frame, self._start_frame): @@ -207,6 +209,8 @@ class LLMFullResponseAggregator(FrameProcessor): self._aggregation = "" async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, TextFrame): self._aggregation += frame.text elif isinstance(frame, LLMFullResponseEndFrame): diff --git a/src/pipecat/processors/aggregators/parallel_task.py b/src/pipecat/processors/aggregators/parallel_task.py index d2142f829..ce341bde5 100644 --- a/src/pipecat/processors/aggregators/parallel_task.py +++ b/src/pipecat/processors/aggregators/parallel_task.py @@ -22,6 +22,8 @@ class Source(FrameProcessor): self._up_queue = upstream_queue async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + match direction: case FrameDirection.UPSTREAM: await self._up_queue.put(frame) @@ -36,6 +38,8 @@ class Sink(FrameProcessor): self._down_queue = downstream_queue async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + match direction: case FrameDirection.UPSTREAM: await self.push_frame(frame, direction) @@ -80,6 +84,8 @@ class ParallelTask(FrameProcessor): # async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if direction == FrameDirection.UPSTREAM: # If we get an upstream frame we process it in each sink. await asyncio.gather(*[s.process_frame(frame, direction) for s in self._sinks]) diff --git a/src/pipecat/processors/aggregators/sentence.py b/src/pipecat/processors/aggregators/sentence.py index 3f323bbb5..a7992eace 100644 --- a/src/pipecat/processors/aggregators/sentence.py +++ b/src/pipecat/processors/aggregators/sentence.py @@ -33,6 +33,8 @@ class SentenceAggregator(FrameProcessor): self._aggregation = "" async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + # We ignore interim description at this point. if isinstance(frame, InterimTranscriptionFrame): return diff --git a/src/pipecat/processors/aggregators/user_response.py b/src/pipecat/processors/aggregators/user_response.py index 5ce62b859..12f3bcb93 100644 --- a/src/pipecat/processors/aggregators/user_response.py +++ b/src/pipecat/processors/aggregators/user_response.py @@ -82,6 +82,8 @@ class ResponseAggregator(FrameProcessor): # and T2 would be dropped. async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + send_aggregation = False if isinstance(frame, self._start_frame): diff --git a/src/pipecat/processors/aggregators/vision_image_frame.py b/src/pipecat/processors/aggregators/vision_image_frame.py index 45fd6756b..f0c8a9c76 100644 --- a/src/pipecat/processors/aggregators/vision_image_frame.py +++ b/src/pipecat/processors/aggregators/vision_image_frame.py @@ -30,6 +30,8 @@ class VisionImageFrameAggregator(FrameProcessor): self._describe_text = None async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, TextFrame): self._describe_text = frame.text elif isinstance(frame, ImageRawFrame): diff --git a/src/pipecat/processors/filters/frame_filter.py b/src/pipecat/processors/filters/frame_filter.py index 42f0e9dd8..9f2eb98c4 100644 --- a/src/pipecat/processors/filters/frame_filter.py +++ b/src/pipecat/processors/filters/frame_filter.py @@ -30,5 +30,7 @@ class FrameFilter(FrameProcessor): or isinstance(frame, SystemFrame)) async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if self._should_passthrough_frame(frame): await self.push_frame(frame, direction) diff --git a/src/pipecat/processors/filters/wake_check_filter.py b/src/pipecat/processors/filters/wake_check_filter.py index 44faa03bf..7704a5732 100644 --- a/src/pipecat/processors/filters/wake_check_filter.py +++ b/src/pipecat/processors/filters/wake_check_filter.py @@ -43,6 +43,8 @@ class WakeCheckFilter(FrameProcessor): self._wake_patterns.append(pattern) async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + try: if isinstance(frame, TranscriptionFrame): p = self._participant_states.get(frame.user_id) diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 793aa8eb1..0b9c71fef 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -8,7 +8,7 @@ import asyncio from enum import Enum -from pipecat.frames.frames import ErrorFrame, Frame +from pipecat.frames.frames import ErrorFrame, Frame, StartFrame from pipecat.utils.utils import obj_count, obj_id from loguru import logger @@ -28,6 +28,18 @@ class FrameProcessor: self._next: "FrameProcessor" | None = None self._loop: asyncio.AbstractEventLoop = loop or asyncio.get_running_loop() + # Properties + self._allow_interruptions = False + self._enable_metrics = False + + @property + def allow_interruptions(self): + return self._allow_interruptions + + @property + def enable_metrics(self): + return self._enable_metrics + async def cleanup(self): pass @@ -40,7 +52,9 @@ class FrameProcessor: return self._loop async def process_frame(self, frame: Frame, direction: FrameDirection): - pass + if isinstance(frame, StartFrame): + self._allow_interruptions = frame.allow_interruptions + self._enable_metrics = frame.enable_metrics async def push_error(self, error: ErrorFrame): await self.push_frame(error, FrameDirection.UPSTREAM) diff --git a/src/pipecat/processors/frameworks/langchain.py b/src/pipecat/processors/frameworks/langchain.py index bac39cf26..674d3aa97 100644 --- a/src/pipecat/processors/frameworks/langchain.py +++ b/src/pipecat/processors/frameworks/langchain.py @@ -39,6 +39,8 @@ class LangchainProcessor(FrameProcessor): self._participant_id = participant_id async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, LLMMessagesFrame): # Messages are accumulated by the `LLMUserResponseAggregator` in a list of messages. # The last one by the human is the one we want to send to the LLM. diff --git a/src/pipecat/processors/text_transformer.py b/src/pipecat/processors/text_transformer.py index 550b8dc1c..39fcf6b67 100644 --- a/src/pipecat/processors/text_transformer.py +++ b/src/pipecat/processors/text_transformer.py @@ -27,6 +27,8 @@ class StatelessTextTransformer(FrameProcessor): self._transform_fn = transform_fn async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, TextFrame): result = self._transform_fn(frame.text) if isinstance(result, Coroutine): diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index fc879d887..a46f41f35 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -106,6 +106,8 @@ class TTSService(AIService): await self.push_frame(TextFrame(text)) async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, TextFrame): await self._process_text_frame(frame) elif isinstance(frame, EndFrame): @@ -179,6 +181,8 @@ class STTService(AIService): async def process_frame(self, frame: Frame, direction: FrameDirection): """Processes a frame of audio data, either buffering or transcribing it.""" + await super().process_frame(frame, direction) + if isinstance(frame, CancelFrame) or isinstance(frame, EndFrame): self._wave.close() await self.push_frame(frame, direction) @@ -201,6 +205,8 @@ class ImageGenService(AIService): pass async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, TextFrame): await self.push_frame(frame, direction) await self.process_generator(self.run_image_gen(frame.text)) @@ -220,6 +226,8 @@ class VisionService(AIService): pass async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, VisionImageRawFrame): await self.process_generator(self.run_vision(frame)) else: diff --git a/src/pipecat/services/anthropic.py b/src/pipecat/services/anthropic.py index 3e774a6bd..956941073 100644 --- a/src/pipecat/services/anthropic.py +++ b/src/pipecat/services/anthropic.py @@ -122,6 +122,8 @@ class AnthropicLLMService(LLMService): await self.push_frame(LLMFullResponseEndFrame()) async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + context = None if isinstance(frame, OpenAILLMContextFrame): diff --git a/src/pipecat/services/google.py b/src/pipecat/services/google.py index 81d5e756e..89ac6d83d 100644 --- a/src/pipecat/services/google.py +++ b/src/pipecat/services/google.py @@ -105,6 +105,8 @@ class GoogleLLMService(LLMService): await self.push_frame(LLMFullResponseEndFrame()) async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + context = None if isinstance(frame, OpenAILLMContextFrame): diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index 4dca1a998..57bcc09d9 100644 --- a/src/pipecat/services/openai.py +++ b/src/pipecat/services/openai.py @@ -215,6 +215,8 @@ class BaseOpenAILLMService(LLMService): raise BaseException(f"Unknown return type from function callback: {type(result)}") async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + context = None if isinstance(frame, OpenAILLMContextFrame): context: OpenAILLMContext = frame.context diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 531269bec..3657727fb 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -34,7 +34,6 @@ class BaseInputTransport(FrameProcessor): self._params = params self._running = False - self._allow_interruptions = False self._executor = ThreadPoolExecutor(max_workers=5) @@ -43,11 +42,6 @@ class BaseInputTransport(FrameProcessor): self._create_push_task() async def start(self, frame: StartFrame): - # Make sure we have the latest params. Note that this transport might - # have been started on another task that might not need interruptions, - # for example. - self._allow_interruptions = frame.allow_interruptions - if self._running: return @@ -86,12 +80,13 @@ class BaseInputTransport(FrameProcessor): pass async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, CancelFrame): # We don't queue a CancelFrame since we want to stop ASAP. await self.push_frame(frame, direction) await self.stop() elif isinstance(frame, StartFrame): - self._allow_interruption = frame.allow_interruptions await self.start(frame) await self._internal_push_frame(frame, direction) elif isinstance(frame, EndFrame): @@ -128,7 +123,7 @@ class BaseInputTransport(FrameProcessor): # async def _handle_interruptions(self, frame: Frame): - if self._allow_interruptions: + if self.allow_interruptions: # Make sure we notify about interruptions quickly out-of-band if isinstance(frame, UserStartedSpeakingFrame): logger.debug("User started speaking") diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index d167919f6..e880c99d4 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -41,7 +41,6 @@ class BaseOutputTransport(FrameProcessor): self._params = params self._running = False - self._allow_interruptions = False self._executor = ThreadPoolExecutor(max_workers=5) @@ -62,11 +61,6 @@ class BaseOutputTransport(FrameProcessor): self._create_push_task() async def start(self, frame: StartFrame): - # Make sure we have the latest params. Note that this transport might - # have been started on another task that might not need interruptions, - # for example. - self._allow_interruptions = frame.allow_interruptions - if self._running: return @@ -111,6 +105,8 @@ class BaseOutputTransport(FrameProcessor): await self._sink_thread async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + # # Out-of-band frames like (CancelFrame or StartInterruptionFrame) are # pushed immediately. Other frames require order so they are put in the @@ -136,7 +132,7 @@ class BaseOutputTransport(FrameProcessor): await self._stopped_event.wait() async def _handle_interruptions(self, frame: Frame): - if not self._allow_interruptions: + if not self.allow_interruptions: return if isinstance(frame, StartInterruptionFrame): diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 893aa9400..8e42a1678 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -521,6 +521,8 @@ class DailyInputTransport(BaseInputTransport): # async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, UserImageRequestFrame): self.request_participant_image(frame.user_id) diff --git a/src/pipecat/utils/test_frame_processor.py b/src/pipecat/utils/test_frame_processor.py index f4a6674aa..2354c40e0 100644 --- a/src/pipecat/utils/test_frame_processor.py +++ b/src/pipecat/utils/test_frame_processor.py @@ -13,6 +13,8 @@ class TestFrameProcessor(FrameProcessor): super().__init__() async def process_frame(self, frame, direction): + await super().process_frame(frame, direction) + if not self.test_frames[0]: # then we've run out of required frames but the generator is still going? raise TestException(f"Oops, got an extra frame, {frame}") if isinstance(self.test_frames[0], List): diff --git a/src/pipecat/vad/silero.py b/src/pipecat/vad/silero.py index 97d0a2144..52ea159b1 100644 --- a/src/pipecat/vad/silero.py +++ b/src/pipecat/vad/silero.py @@ -94,6 +94,8 @@ class SileroVAD(FrameProcessor): # async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, AudioRawFrame): await self._analyze_audio(frame) if self._audio_passthrough: diff --git a/tests/test_langchain.py b/tests/test_langchain.py index 6b37d0987..7b32b2a9a 100644 --- a/tests/test_langchain.py +++ b/tests/test_langchain.py @@ -36,6 +36,8 @@ class TestLangchain(unittest.IsolatedAsyncioTestCase): return self.name async def process_frame(self, frame, direction): + await super().process_frame(frame, direction) + if isinstance(frame, LLMFullResponseStartFrame): self.start_collecting = True elif isinstance(frame, TextFrame) and self.start_collecting: From 390582d7f3b558449968402bae2544ce2efecf6c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 6 Jun 2024 13:51:59 -0700 Subject: [PATCH 09/10] services: use start/stop_ttfb_metrics to report TTFB metrics --- CHANGELOG.md | 11 +++++++++-- src/pipecat/frames/frames.py | 7 +++++++ src/pipecat/pipeline/task.py | 9 +++++++-- src/pipecat/processors/frame_processor.py | 21 ++++++++++++++++++--- src/pipecat/services/anthropic.py | 8 +++++--- src/pipecat/services/azure.py | 10 +++------- src/pipecat/services/cartesia.py | 9 +++------ src/pipecat/services/deepgram.py | 8 ++------ src/pipecat/services/elevenlabs.py | 9 +++------ src/pipecat/services/google.py | 14 +++++++++----- src/pipecat/services/openai.py | 20 ++++++++++---------- src/pipecat/services/playht.py | 14 ++++---------- src/pipecat/services/whisper.py | 3 +++ src/pipecat/transports/base_input.py | 2 +- src/pipecat/transports/base_output.py | 2 +- 15 files changed, 85 insertions(+), 62 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a9b78496f..2b7b3d7f4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,11 +10,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added -- Added TTFB debug logging for TTS services +- Added `enable_metrics` to `PipelineParams`. + +- Added `MetricsFrame`. The `MetricsFrame` will report different metrics in the + system. Right now, it can report TTFB (Time To First Byte) values for + different services, that is the time spent between the arrival of a `Frame` to + the processor/service until the first `DataFrame` is pushed downstream. + +- Added TTFB metrics and debug logging for TTS services. ### Fixed -- Fixed PlayHT TTS service to work properly async +- Fixed PlayHT TTS service to work properly async. ## [0.0.28] - 2024-06-05 diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 9a5725de1..b535d0de0 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -239,6 +239,13 @@ class StopInterruptionFrame(SystemFrame): pass +@dataclass +class MetricsFrame(SystemFrame): + """Emitted by processor who can compute metrics like latencies. + """ + ttfb: Mapping[str, float] + + # # Control frames # diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index b1f72c096..76d98165b 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -19,6 +19,7 @@ from loguru import logger class PipelineParams(BaseModel): allow_interruptions: bool = False + enable_metrics: bool = False class Source(FrameProcessor): @@ -89,8 +90,12 @@ class PipelineTask: raise Exception("Frames must be an iterable or async iterable") async def _process_down_queue(self): - await self._source.process_frame( - StartFrame(allow_interruptions=self._params.allow_interruptions), FrameDirection.DOWNSTREAM) + start_frame = StartFrame( + allow_interruptions=self._params.allow_interruptions, + enable_metrics=self._params.enable_metrics, + ) + await self._source.process_frame(start_frame, FrameDirection.DOWNSTREAM) + running = True should_cleanup = True while running: diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 0b9c71fef..e0a072477 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -5,10 +5,11 @@ # import asyncio +import time from enum import Enum -from pipecat.frames.frames import ErrorFrame, Frame, StartFrame +from pipecat.frames.frames import ErrorFrame, Frame, MetricsFrame, StartFrame from pipecat.utils.utils import obj_count, obj_id from loguru import logger @@ -32,14 +33,28 @@ class FrameProcessor: self._allow_interruptions = False self._enable_metrics = False + # Metrics + self._start_ttfb_time = 0 + @property - def allow_interruptions(self): + def interruptions_allowed(self): return self._allow_interruptions @property - def enable_metrics(self): + def metrics_enabled(self): return self._enable_metrics + async def start_ttfb_metrics(self): + if self.metrics_enabled: + self._start_ttfb_time = time.time() + + async def stop_ttfb_metrics(self): + if self.metrics_enabled and self._start_ttfb_time > 0: + ttfb = time.time() - self._start_ttfb_time + logger.debug(f"{self.name} TTFB: {ttfb}") + await self.push_frame(MetricsFrame(ttfb={self.name: ttfb})) + self._start_ttfb_time = 0 + async def cleanup(self): pass diff --git a/src/pipecat/services/anthropic.py b/src/pipecat/services/anthropic.py index 956941073..95b31f336 100644 --- a/src/pipecat/services/anthropic.py +++ b/src/pipecat/services/anthropic.py @@ -4,7 +4,6 @@ # SPDX-License-Identifier: BSD 2-Clause License # -import time import base64 from pipecat.frames.frames import ( @@ -102,13 +101,16 @@ class AnthropicLLMService(LLMService): messages = self._get_messages_from_openai_context(context) - start_time = time.time() + await self.start_ttfb_metric() + response = await self._client.messages.create( messages=messages, model=self._model, max_tokens=self._max_tokens, stream=True) - logger.debug(f"Anthropic LLM TTFB: {time.time() - start_time}") + + await self.stop_ttfb_metric() + async for event in response: # logger.debug(f"Anthropic LLM event: {event}") if (event.type == "content_block_delta"): diff --git a/src/pipecat/services/azure.py b/src/pipecat/services/azure.py index fce3c5939..d35584303 100644 --- a/src/pipecat/services/azure.py +++ b/src/pipecat/services/azure.py @@ -7,12 +7,10 @@ import aiohttp import asyncio import io -import time from PIL import Image from typing import AsyncGenerator -from numpy import str_ from openai import AsyncAzureOpenAI from pipecat.frames.frames import AudioRawFrame, ErrorFrame, Frame, URLImageRawFrame @@ -47,10 +45,10 @@ class AzureTTSService(TTSService): self._voice = voice async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - start_time = time.time() - ttfb = None logger.debug(f"Generating TTS: {text}") + await self.start_ttfb_metrics() + ssml = ( "" @@ -64,9 +62,7 @@ class AzureTTSService(TTSService): result = await asyncio.to_thread(self.speech_synthesizer.speak_ssml, (ssml)) if result.reason == ResultReason.SynthesizingAudioCompleted: - if ttfb is None: - ttfb = time.time() - start_time - logger.debug(f"TTS ttfb: {ttfb}") + await self.stop_ttfb_metrics() # Azure always sends a 44-byte header. Strip it off. yield AudioRawFrame(audio=result.audio_data[44:], sample_rate=16000, num_channels=1) elif result.reason == ResultReason.Canceled: diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index 9c730a283..474600ffa 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -3,7 +3,6 @@ # # SPDX-License-Identifier: BSD 2-Clause License # -import time from cartesia.tts import AsyncCartesiaTTS @@ -41,11 +40,11 @@ class CartesiaTTSService(TTSService): logger.error(f"Cartesia initialization error: {e}") async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - start_time = time.time() - ttfb = None logger.debug(f"Generating TTS: [{text}]") try: + await self.start_ttfb_metrics() + chunk_generator = await self._client.generate( stream=True, transcript=text, @@ -55,9 +54,7 @@ class CartesiaTTSService(TTSService): ) async for chunk in chunk_generator: - if ttfb is None: - ttfb = time.time() - start_time - logger.debug(f"TTS ttfb: {ttfb}") + await self.stop_ttfb_metrics() yield AudioRawFrame(chunk["audio"], chunk["sampling_rate"], 1) except Exception as e: logger.error(f"Cartesia exception: {e}") diff --git a/src/pipecat/services/deepgram.py b/src/pipecat/services/deepgram.py index 7b19e04e2..aaf0c01ee 100644 --- a/src/pipecat/services/deepgram.py +++ b/src/pipecat/services/deepgram.py @@ -5,7 +5,6 @@ # import aiohttp -import time from typing import AsyncGenerator @@ -31,8 +30,6 @@ class DeepgramTTSService(TTSService): self._aiohttp_session = aiohttp_session async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - start_time = time.time() - ttfb = None logger.debug(f"Generating TTS: [{text}]") base_url = "https://api.deepgram.com/v1/speak" @@ -41,6 +38,7 @@ class DeepgramTTSService(TTSService): body = {"text": text} try: + await self.start_ttfb_metrics() async with self._aiohttp_session.post(request_url, headers=headers, json=body) as r: if r.status != 200: text = await r.text() @@ -49,9 +47,7 @@ class DeepgramTTSService(TTSService): return async for data in r.content: - if ttfb is None: - ttfb = time.time() - start_time - logger.debug(f"TTS ttfb: {ttfb}") + await self.stop_ttfb_metrics() frame = AudioRawFrame(audio=data, sample_rate=16000, num_channels=1) yield frame except Exception as e: diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index d5b476160..717460fd3 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -5,7 +5,6 @@ # import aiohttp -import time from typing import AsyncGenerator @@ -33,8 +32,6 @@ class ElevenLabsTTSService(TTSService): self._model = model async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - start_time = time.time() - ttfb = None logger.debug(f"Generating TTS: [{text}]") url = f"https://api.elevenlabs.io/v1/text-to-speech/{self._voice_id}/stream" @@ -50,6 +47,8 @@ class ElevenLabsTTSService(TTSService): "Content-Type": "application/json", } + 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() @@ -59,8 +58,6 @@ class ElevenLabsTTSService(TTSService): async for chunk in r.content: if len(chunk) > 0: - if ttfb is None: - ttfb = time.time() - start_time - logger.debug(f"TTS ttfb: {ttfb}") + await self.stop_ttfb_metrics() frame = AudioRawFrame(chunk, 16000, 1) yield frame diff --git a/src/pipecat/services/google.py b/src/pipecat/services/google.py index 89ac6d83d..ffe9a9fee 100644 --- a/src/pipecat/services/google.py +++ b/src/pipecat/services/google.py @@ -1,8 +1,10 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# -import json -import os import asyncio -import time from typing import List @@ -81,9 +83,11 @@ class GoogleLLMService(LLMService): messages = self._get_messages_from_openai_context(context) - start_time = time.time() + await self.start_ttfb_metrics() + response = self._client.generate_content(messages, stream=True) - logger.debug(f"Google LLM TTFB: {time.time() - start_time}") + + await self.stop_ttfb_metrics() async for chunk in self._async_generator_wrapper(response): try: diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index 57bcc09d9..aaed7d8e7 100644 --- a/src/pipecat/services/openai.py +++ b/src/pipecat/services/openai.py @@ -3,13 +3,14 @@ # # SPDX-License-Identifier: BSD 2-Clause License # + +import aiohttp import base64 import io import json -import time + from typing import AsyncGenerator, List, Literal -import aiohttp from loguru import logger from PIL import Image @@ -94,7 +95,6 @@ class BaseOpenAILLMService(LLMService): del message["data"] del message["mime_type"] - start_time = time.time() chunks: AsyncStream[ChatCompletionChunk] = ( await self._client.chat.completions.create( model=self._model, @@ -105,8 +105,6 @@ class BaseOpenAILLMService(LLMService): ) ) - logger.debug(f"OpenAI LLM TTFB: {time.time() - start_time}") - return chunks async def _chat_completions(self, messages) -> str | None: @@ -123,6 +121,8 @@ class BaseOpenAILLMService(LLMService): arguments = "" tool_call_id = "" + await self.start_ttfb_metrics() + chunk_stream: AsyncStream[ChatCompletionChunk] = ( await self._stream_chat_completions(context) ) @@ -131,6 +131,8 @@ class BaseOpenAILLMService(LLMService): if len(chunk.choices) == 0: continue + await self.stop_ttfb_metrics() + if chunk.choices[0].delta.tool_calls: # We're streaming the LLM response to enable the fastest response times. # For text, we just yield each chunk as we receive it and count on consumers @@ -306,11 +308,11 @@ class OpenAITTSService(TTSService): self._client = AsyncOpenAI(api_key=api_key) async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - start_time = time.time() - ttfb = None logger.debug(f"Generating TTS: [{text}]") try: + await self.start_ttfb_metrics() + async with self._client.audio.speech.with_streaming_response.create( input=text, model=self._model, @@ -324,9 +326,7 @@ class OpenAITTSService(TTSService): return async for chunk in r.iter_bytes(8192): if len(chunk) > 0: - if ttfb is None: - ttfb = time.time() - start_time - logger.debug(f"TTS ttfb: {ttfb}") + await self.stop_ttfb_metrics() frame = AudioRawFrame(chunk, 24_000, 1) yield frame except BadRequestError as e: diff --git a/src/pipecat/services/playht.py b/src/pipecat/services/playht.py index 8373eca7b..cf38e125d 100644 --- a/src/pipecat/services/playht.py +++ b/src/pipecat/services/playht.py @@ -6,8 +6,6 @@ import io import struct -import time -import asyncio from typing import AsyncGenerator @@ -49,21 +47,19 @@ class PlayHTTTSService(TTSService): self._client.close() async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - start_time = time.time() - ttfb = None logger.debug(f"Generating TTS: [{text}]") try: b = bytearray() in_header = True + + await self.start_ttfb_metrics() + playht_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 - async for chunk in playht_gen: # skip the RIFF header. if in_header: @@ -80,9 +76,7 @@ class PlayHTTTSService(TTSService): in_header = False else: if len(chunk): - if ttfb is None: - ttfb = time.time() - start_time - logger.debug(f"TTS ttfb: {ttfb}") + await self.stop_ttfb_metrics() frame = AudioRawFrame(chunk, 16000, 1) yield frame except Exception as e: diff --git a/src/pipecat/services/whisper.py b/src/pipecat/services/whisper.py index 7884d454b..6c5dd0d1f 100644 --- a/src/pipecat/services/whisper.py +++ b/src/pipecat/services/whisper.py @@ -73,6 +73,8 @@ class WhisperSTTService(STTService): logger.error("Whisper model not available") return + await self.start_ttfb_metrics() + # Divide by 32768 because we have signed 16-bit data. audio_float = np.frombuffer(audio, dtype=np.int16).astype(np.float32) / 32768.0 @@ -83,4 +85,5 @@ class WhisperSTTService(STTService): text += f"{segment.text} " if text: + await self.stop_ttfb_metrics() yield TranscriptionFrame(text, "", int(time.time_ns() / 1000000)) diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 3657727fb..823ab5844 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -123,7 +123,7 @@ class BaseInputTransport(FrameProcessor): # async def _handle_interruptions(self, frame: Frame): - if self.allow_interruptions: + if self.interruptions_allowed: # Make sure we notify about interruptions quickly out-of-band if isinstance(frame, UserStartedSpeakingFrame): logger.debug("User started speaking") diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index e880c99d4..b81f2f8db 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -132,7 +132,7 @@ class BaseOutputTransport(FrameProcessor): await self._stopped_event.wait() async def _handle_interruptions(self, frame: Frame): - if not self.allow_interruptions: + if not self.interruptions_allowed: return if isinstance(frame, StartInterruptionFrame): From a1f1d1995c5a50af14f4afa7ea549a5d0e5cf539 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 6 Jun 2024 14:06:01 -0700 Subject: [PATCH 10/10] transports: allow sending metrics --- src/pipecat/frames/frames.py | 2 +- src/pipecat/transports/base_output.py | 6 ++++++ src/pipecat/transports/services/daily.py | 13 ++++++++++++- 3 files changed, 19 insertions(+), 2 deletions(-) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index b535d0de0..572246c6d 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -241,7 +241,7 @@ class StopInterruptionFrame(SystemFrame): @dataclass class MetricsFrame(SystemFrame): - """Emitted by processor who can compute metrics like latencies. + """Emitted by processor that can compute metrics like latencies. """ ttfb: Mapping[str, float] diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index b81f2f8db..4c19a0b55 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -20,6 +20,7 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.frames.frames import ( AudioRawFrame, CancelFrame, + MetricsFrame, SpriteFrame, StartFrame, EndFrame, @@ -87,6 +88,9 @@ class BaseOutputTransport(FrameProcessor): def send_message(self, frame: TransportMessageFrame): pass + def send_metrics(self, frame: MetricsFrame): + pass + def write_frame_to_camera(self, frame: ImageRawFrame): pass @@ -166,6 +170,8 @@ class BaseOutputTransport(FrameProcessor): self._set_camera_images(frame.images) elif isinstance(frame, TransportMessageFrame): self.send_message(frame) + elif isinstance(frame, MetricsFrame): + self.send_metrics(frame) else: future = asyncio.run_coroutine_threadsafe( self._internal_push_frame(frame), self.get_event_loop()) diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 8e42a1678..0b440181c 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -27,6 +27,7 @@ from pipecat.frames.frames import ( Frame, ImageRawFrame, InterimTranscriptionFrame, + MetricsFrame, SpriteFrame, StartFrame, TranscriptionFrame, @@ -638,6 +639,16 @@ class DailyOutputTransport(BaseOutputTransport): def send_message(self, frame: DailyTransportMessageFrame): self._client.send_message(frame) + def send_metrics(self, frame: MetricsFrame): + ttfb = [{"name": n, "time": t} for n, t in frame.ttfb.items()] + message = DailyTransportMessageFrame(message={ + "type": "pipecat-metrics", + "metrics": { + "ttfb": ttfb + }, + }) + self._client.send_message(message) + def write_raw_audio_frames(self, frames: bytes): self._client.write_raw_audio_frames(frames) @@ -711,7 +722,7 @@ class DailyTransport(BaseTransport): # DailyTransport # - @property + @ property def participant_id(self) -> str: return self._client.participant_id