wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #217 from pipecat-ai/khk-tts-timings

Browse Source 
Added TTFB timings for all TTS services
This commit is contained in:
Aleix Conchillo Flaqué
2024-06-07 05:42:52 +08:00
committed by GitHub
parent cd60a84f8a a1f1d1995c
commit c25f6e56e7
50 changed files with 555 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
CHANGELOG.md
View File

@@ -5,6 +5,24 @@ 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 `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.
## [0.0.28] - 2024-06-05
### Fixed

2
examples/foundational/05-sync-speech-and-image.py
View File

@@ -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):

6
examples/foundational/05a-local-sync-speech-and-image.py
View File

@@ -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

2
examples/foundational/06a-image-sync.py
View File

@@ -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)

1
examples/foundational/07d-interruptible-cartesia.py
View File

@@ -20,6 +20,7 @@ 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

96
examples/foundational/07e-interruptible-playht.py Normal file
View File

@@ -0,0 +1,96 @@
#
# 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
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))

95
examples/foundational/07f-interruptible-azure-tts.py Normal file
View File

@@ -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))

94
examples/foundational/07g-interruptible-openai-tts.py Normal file
View File

@@ -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))

4
examples/foundational/11-sound-effects.py
View File

@@ -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

2
examples/foundational/12-describe-video.py
View File

@@ -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)

2
examples/foundational/12a-describe-video-gemini-flash.py
View File

@@ -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)

2
examples/foundational/12b-describe-video-gpt-4o.py
View File

@@ -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)

2
examples/foundational/12c-describe-video-anthropic.py
View File

@@ -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)

2
examples/foundational/13-whisper-transcription.py
View File

@@ -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}")

2
examples/foundational/13a-whisper-local.py
View File

@@ -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}")

8
examples/moondream-chatbot/bot.py
View File

@@ -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)

2
examples/simple-chatbot/bot.py
View File

@@ -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)

4
examples/storytelling-chatbot/src/processors.py
View File

@@ -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))

4
examples/translation-chatbot/bot.py
View File

@@ -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,

10
src/pipecat/frames/frames.py
View File

@@ -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
@@ -238,6 +239,13 @@ class StopInterruptionFrame(SystemFrame):
pass
@dataclass
class MetricsFrame(SystemFrame):
"""Emitted by processor that can compute metrics like latencies.
"""
ttfb: Mapping[str, float]
#
# Control frames
#

6
src/pipecat/pipeline/parallel_pipeline.py
View File

@@ -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()

6
src/pipecat/pipeline/pipeline.py
View File

@@ -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:

11
src/pipecat/pipeline/task.py
View File

@@ -19,6 +19,7 @@ from loguru import logger
class PipelineParams(BaseModel):
allow_interruptions: bool = False
enable_metrics: bool = False
class Source(FrameProcessor):
@@ -28,6 +29,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)
@@ -87,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:

2
src/pipecat/processors/aggregators/gated.py
View File

@@ -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)

4
src/pipecat/processors/aggregators/llm_response.py
View File

@@ -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):

6
src/pipecat/processors/aggregators/parallel_task.py
View File

@@ -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])

2
src/pipecat/processors/aggregators/sentence.py
View File

@@ -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

2
src/pipecat/processors/aggregators/user_response.py
View File

@@ -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):

2
src/pipecat/processors/aggregators/vision_image_frame.py
View File

@@ -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):

2
src/pipecat/processors/filters/frame_filter.py
View File

@@ -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)

2
src/pipecat/processors/filters/wake_check_filter.py
View File

@@ -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)

33
src/pipecat/processors/frame_processor.py
View File

@@ -5,10 +5,11 @@
#
import asyncio
import time
from enum import Enum
from pipecat.frames.frames import ErrorFrame, Frame
from pipecat.frames.frames import ErrorFrame, Frame, MetricsFrame, StartFrame
from pipecat.utils.utils import obj_count, obj_id
from loguru import logger
@@ -28,6 +29,32 @@ 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
# Metrics
self._start_ttfb_time = 0
@property
def interruptions_allowed(self):
return self._allow_interruptions
@property
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
@@ -40,7 +67,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)

2
src/pipecat/processors/frameworks/langchain.py
View File

@@ -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.

2
src/pipecat/processors/text_transformer.py
View File

@@ -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):

8
src/pipecat/services/ai_services.py
View File

@@ -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:

10
src/pipecat/services/anthropic.py
View File

@@ -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"):
@@ -122,6 +124,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):

4
src/pipecat/services/azure.py
View File

@@ -11,7 +11,6 @@ import io
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
@@ -48,6 +47,8 @@ class AzureTTSService(TTSService):
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
logger.debug(f"Generating TTS: {text}")
await self.start_ttfb_metrics()
ssml = (
"<speak version='1.0' xml:lang='en-US' xmlns='http://www.w3.org/2001/10/synthesis' "
"xmlns:mstts='http://www.w3.org/2001/mstts'>"
@@ -61,6 +62,7 @@ class AzureTTSService(TTSService):
result = await asyncio.to_thread(self.speech_synthesizer.speak_ssml, (ssml))
if result.reason == ResultReason.SynthesizingAudioCompleted:
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:

3
src/pipecat/services/cartesia.py
View File

@@ -43,6 +43,8 @@ class CartesiaTTSService(TTSService):
logger.debug(f"Generating TTS: [{text}]")
try:
await self.start_ttfb_metrics()
chunk_generator = await self._client.generate(
stream=True,
transcript=text,
@@ -52,6 +54,7 @@ class CartesiaTTSService(TTSService):
)
async for chunk in chunk_generator:
await self.stop_ttfb_metrics()
yield AudioRawFrame(chunk["audio"], chunk["sampling_rate"], 1)
except Exception as e:
logger.error(f"Cartesia exception: {e}")

2
src/pipecat/services/deepgram.py
View File

@@ -38,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()
@@ -46,6 +47,7 @@ class DeepgramTTSService(TTSService):
return
async for data in r.content:
await self.stop_ttfb_metrics()
frame = AudioRawFrame(audio=data, sample_rate=16000, num_channels=1)
yield frame
except Exception as e:

3
src/pipecat/services/elevenlabs.py
View File

@@ -47,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()
@@ -56,5 +58,6 @@ class ElevenLabsTTSService(TTSService):
async for chunk in r.content:
if len(chunk) > 0:
await self.stop_ttfb_metrics()
frame = AudioRawFrame(chunk, 16000, 1)
yield frame

16
src/pipecat/services/google.py
View File

@@ -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:
@@ -105,6 +109,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):

17
src/pipecat/services/openai.py
View File

@@ -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
@@ -215,6 +217,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
@@ -307,6 +311,8 @@ class OpenAITTSService(TTSService):
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,
@@ -320,6 +326,7 @@ class OpenAITTSService(TTSService):
return
async for chunk in r.iter_bytes(8192):
if len(chunk) > 0:
await self.stop_ttfb_metrics()
frame = AudioRawFrame(chunk, 24_000, 1)
yield frame
except BadRequestError as e:

63
src/pipecat/services/playht.py
View File

@@ -15,8 +15,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}")
@@ -25,7 +25,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)
@@ -33,7 +33,7 @@ class PlayHTAIService(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,
)
@@ -47,28 +47,37 @@ 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)
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)
async for chunk in playht_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):
await self.stop_ttfb_metrics()
frame = AudioRawFrame(chunk, 16000, 1)
yield frame
except Exception as e:
logger.error(f"Error generating TTS: {e}")

3
src/pipecat/services/whisper.py
View File

@@ -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))

11
src/pipecat/transports/base_input.py
View File

@@ -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.interruptions_allowed:
# Make sure we notify about interruptions quickly out-of-band
if isinstance(frame, UserStartedSpeakingFrame):
logger.debug("User started speaking")

16
src/pipecat/transports/base_output.py
View File

@@ -20,6 +20,7 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.frames.frames import (
AudioRawFrame,
CancelFrame,
MetricsFrame,
SpriteFrame,
StartFrame,
EndFrame,
@@ -41,7 +42,6 @@ class BaseOutputTransport(FrameProcessor):
self._params = params
self._running = False
self._allow_interruptions = False
self._executor = ThreadPoolExecutor(max_workers=5)
@@ -62,11 +62,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
@@ -93,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
@@ -111,6 +109,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 +136,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):
@@ -170,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())

15
src/pipecat/transports/services/daily.py
View File

@@ -27,6 +27,7 @@ from pipecat.frames.frames import (
Frame,
ImageRawFrame,
InterimTranscriptionFrame,
MetricsFrame,
SpriteFrame,
StartFrame,
TranscriptionFrame,
@@ -521,6 +522,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)
@@ -636,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)
@@ -709,7 +722,7 @@ class DailyTransport(BaseTransport):
# DailyTransport
#
@property
@ property
def participant_id(self) -> str:
return self._client.participant_id

2
src/pipecat/utils/test_frame_processor.py
View File

@@ -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):

2
src/pipecat/vad/silero.py
View File

@@ -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:

2
tests/test_langchain.py
View File

@@ -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: