diff --git a/examples/fast-bot-metrics/env-example b/examples/fast-bot-metrics/env-example new file mode 100644 index 000000000..69aba2f99 --- /dev/null +++ b/examples/fast-bot-metrics/env-example @@ -0,0 +1,15 @@ +DAILY_SAMPLE_ROOM_URL=... +DAILY_API_KEY=... + +DEEPGRAM_API_KEY=... +DEEPGRAM_VOICE=alpha-asteria-en-v2 +# DEEPGRAM_STT_URL=ws://localhost:8080 +# DEEPGRAM_TTS_BASE_URL=http://0.0.0.0:8080/v1/speak + +OPENAI_API_KEY=... +# OPENAI_API_KEY= +OPENAI_MODEL=gpt-4o +# OPENAI_MODEL=meta-llama/Meta-Llama-3-70B-Instruct +# OPENAI_MODEL=meta-llama/Meta-Llama-3-8B-Instruct +# OPENAI_MODEL=neuralmagic/Meta-Llama-3-70B-Instruct-FP8 +# OPENAI_BASE_URL=http://0.0.0.0:8000/v1 diff --git a/examples/fast-bot-metrics/fastbothelpers.py b/examples/fast-bot-metrics/fastbothelpers.py new file mode 100644 index 000000000..bfa115e68 --- /dev/null +++ b/examples/fast-bot-metrics/fastbothelpers.py @@ -0,0 +1,268 @@ +from loguru import logger +import asyncio +import math +import struct +import time +from dataclasses import dataclass, field +from typing import List + + +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.frames.frames import ( + Frame, + AudioRawFrame, + InterimTranscriptionFrame, + TranscriptionFrame, + TextFrame, + StartInterruptionFrame, + StopInterruptionFrame, + LLMFullResponseStartFrame, + TTSStoppedFrame, + MetricsFrame +) + +from pipecat.vad.vad_analyzer import VADAnalyzer, VADState +from pipecat.services.deepgram import DeepgramTTSService, DeepgramSTTService +from pipecat.services.openai import OpenAILLMContext, OpenAILLMContextFrame + + +class GreedyLLMAggregator(FrameProcessor): + def __init__(self, context: OpenAILLMContext = None, **kwargs): + super().__init__(**kwargs) + self.context: OpenAILLMContext = context if context else OpenAILLMContext() + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + logger.debug(f"{frame}") + + try: + if isinstance(frame, InterimTranscriptionFrame): + return + + if isinstance(frame, TranscriptionFrame): + # append transcribed text to last "user" frame + if self.context.messages and self.context.messages[-1]["role"] == "user": + last_frame = self.context.messages.pop() + else: + last_frame = {"role": "user", "content": ""} + + last_frame["content"] += " " + frame.text + self.context.messages.append(last_frame) + + oai_context_frame = OpenAILLMContextFrame(context=self.context) + logger.debug(f"pushing frame {oai_context_frame}") + await self.push_frame(oai_context_frame) + return + + await self.push_frame(frame, direction) + except Exception as e: + logger.debug(f"error: {e}") + + +class ClearableDeepgramTTSService(DeepgramTTSService): + def __init___(self, **kwargs): + super().__init(**kwargs) + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if isinstance(frame, StartInterruptionFrame): + self._current_sentence = "" + + +@dataclass +class BufferedSentence: + audio_frames: List[AudioRawFrame] = field(default_factory=list) + text_frame: TextFrame = None + + +class VADGate(FrameProcessor): + + def __init__( + self, + vad_analyzer: VADAnalyzer = None, + context: OpenAILLMContext = None, + **kwargs): + super().__init__(**kwargs) + self.vad_analyzer = vad_analyzer + self.context = context + + self._audio_pusher_task = None + self._expect_text_frame_next = False + self._sentences: List[BufferedSentence] = [] + + # queue output from tts one sentence at a time. associate a buffer of audio frames with the content of + # each text frame. + # + # start a coroutine to service the queue and send sentences down the pipeline when possible. + # 1. do not send anything when we are not in VADState.QUIET + # 2. if we are in VADState.QUIET, send a sentence, estimate how long it will take for that sentence + # to output, sleep until it's time to send another sentence + # 3. each time we send a sentence, append it to the conversation context + # 3. when the sentence buffer becomes empty, cancel the coroutine + # 4. if we get a new LLMFullResponse, treat that as a cancellation, too + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + try: + + # A TTSService will emit a series of AudioRawFrame objects, then a TTSStoppedFrame, + # then a TextFrame. + + if self._expect_text_frame_next: + self._expect_text_frame_next = False + if isinstance(frame, TextFrame): + self._sentences[-1].text_frame = frame + else: + logger.debug(f"expected a text frame, but received {frame}") + await self.push_frame(frame, direction) + return + else: + if isinstance(frame, TextFrame): + logger.error(f"XXXXXXXXXXXXXXXXXXX received a text frame, wasn't expecting it.") + + if isinstance(frame, AudioRawFrame): + # if our buffer is empty or has a "finished" sentence at the end, + # then we need to start buffering a new sentence + if not self._sentences or self._sentences[-1].text_frame: + self._sentences.append(BufferedSentence()) + self._sentences[-1].audio_frames.append(frame) + await self.maybe_start_audio_pusher_task() + return + + if isinstance(frame, TTSStoppedFrame): + self._expect_text_frame_next = True + await self.push_frame(frame, direction) + return + + # There are two ways we can be interrupted. During greedy inference, a new + # LLM response can start. Or, during playout, we can get a traditional + # user interruption frame. + if (isinstance(frame, LLMFullResponseStartFrame) or + isinstance(frame, StartInterruptionFrame)): + logger.debug(f"{frame} - Handle interruption in VADGate") + self._sentences = [] + if self._audio_pusher_task: + self._audio_pusher_task.cancel() + self._audio_pusher_task = None + await self.push_frame(frame, direction) + return + + await self.push_frame(frame, direction) + except Exception as e: + logger.debug(f"error: {e}") + + async def maybe_start_audio_pusher_task(self): + try: + if self._audio_pusher_task: + return + self._audio_pusher_task = self.get_event_loop().create_task(self.push_audio()) + + except Exception as e: + logger.debug(f"Exception {e}") + + async def push_audio(self): + try: + while True: + if not self._sentences: + await asyncio.sleep(0.01) + continue + + if self.vad_analyzer._vad_state != VADState.QUIET: + await asyncio.sleep(0.01) + continue + + # we only want to push completed sentence buffers + if not self._sentences[0].text_frame: + await asyncio.sleep(0.01) + continue + + s = self._sentences.pop(0) + if not s.audio_frames: + continue + sample_rate = s.audio_frames[0].sample_rate + duration = 0 + logger.debug(f"Pushing {len(s.audio_frames)} audio frames") + for frame in s.audio_frames: + await self.push_frame(frame) + # assume linear16 encoding (2 bytes per sample). todo: add some more + # metadata to AudioRawFrame, maybe + duration += (len(frame.audio) / 2 / frame.num_channels) / sample_rate + await asyncio.sleep(duration - 20 / 1000) + if self.context: + logger.debug(f"Appending assistant message to context: [{s.text_frame.text}]") + self.context.messages.append( + {"role": "assistant", "content": s.text_frame.text} + ) + await self.push_frame(s.text_frame) + + except Exception as e: + logger.debug(f"Exception {e}") + + +class TranscriptionTimingLogger(FrameProcessor): + def __init__(self, avt): + super().__init__() + self.name = "Transcription" + self._avt = avt + + async def process_frame(self, frame: Frame, direction: FrameDirection): + try: + await super().process_frame(frame, direction) + if isinstance(frame, TranscriptionFrame): + elapsed = time.time() - self._avt.last_transition_ts + logger.debug(f"Transcription TTF: {elapsed}") + await self.push_frame(MetricsFrame(ttfb={self.name: elapsed})) + + await self.push_frame(frame, direction) + except Exception as e: + logger.debug(f"Exception {e}") + + +class AudioVolumeTimer(FrameProcessor): + def __init__(self): + super().__init__() + self.last_transition_ts = 0 + self._prev_volume = -80 + self._speech_volume_threshold = -50 + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if isinstance(frame, AudioRawFrame): + volume = self.calculate_volume(frame) + # print(f"Audio volume: {volume:.2f} dB") + if (volume >= self._speech_volume_threshold and + self._prev_volume < self._speech_volume_threshold): + # logger.debug("transition above speech volume threshold") + self.last_transition_ts = time.time() + elif (volume < self._speech_volume_threshold and + self._prev_volume >= self._speech_volume_threshold): + # logger.debug("transition below non-speech volume threshold") + self.last_transition_ts = time.time() + self._prev_volume = volume + + await self.push_frame(frame, direction) + + def calculate_volume(self, frame: AudioRawFrame) -> float: + if frame.num_channels != 1: + raise ValueError(f"Expected 1 channel, got {frame.num_channels}") + + # Unpack audio data into 16-bit integers + fmt = f"{len(frame.audio)//2}h" + audio_samples = struct.unpack(fmt, frame.audio) + + # Calculate RMS + sum_squares = sum(sample**2 for sample in audio_samples) + rms = math.sqrt(sum_squares / len(audio_samples)) + + # Convert RMS to decibels (dB) + # Reference: maximum value for 16-bit audio is 32767 + if rms > 0: + db = 20 * math.log10(rms / 32767) + else: + db = -96 # Minimum value (almost silent) + + return db diff --git a/examples/fast-bot-metrics/runner.py b/examples/fast-bot-metrics/runner.py new file mode 100644 index 000000000..6d1a8113d --- /dev/null +++ b/examples/fast-bot-metrics/runner.py @@ -0,0 +1,58 @@ +import argparse +import os +import time +import urllib +import requests + + +def configure(): + parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample") + parser.add_argument( + "-u", + "--url", + type=str, + required=False, + help="URL of the Daily room to join") + parser.add_argument( + "-k", + "--apikey", + type=str, + required=False, + help="Daily API Key (needed to create an owner token for the room)", + ) + + args, unknown = parser.parse_known_args() + + url = args.url or os.getenv("DAILY_SAMPLE_ROOM_URL") + key = args.apikey or os.getenv("DAILY_API_KEY") + + if not url: + raise Exception( + "No Daily room specified. use the -u/--url option from the command line, or set DAILY_SAMPLE_ROOM_URL in your environment to specify a Daily room URL.") + + if not key: + raise Exception("No Daily API key specified. use the -k/--apikey option from the command line, or set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers.") + + # Create a meeting token for the given room with an expiration 1 hour in + # the future. + room_name: str = urllib.parse.urlparse(url).path[1:] + expiration: float = time.time() + 60 * 60 + + res: requests.Response = requests.post( + f"https://api.daily.co/v1/meeting-tokens", + headers={ + "Authorization": f"Bearer {key}"}, + json={ + "properties": { + "room_name": room_name, + "is_owner": True, + "exp": expiration}}, + ) + + if res.status_code != 200: + raise Exception( + f"Failed to create meeting token: {res.status_code} {res.text}") + + token: str = res.json()["token"] + + return (url, token) diff --git a/examples/fast-bot-metrics/tmp.py b/examples/fast-bot-metrics/tmp.py new file mode 100644 index 000000000..3dafee4bc --- /dev/null +++ b/examples/fast-bot-metrics/tmp.py @@ -0,0 +1,165 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +from loguru import logger +from runner import configure +import asyncio +import aiohttp +import os +import sys +from typing import List + + +from pipecat.vad.vad_analyzer import VADParams +from pipecat.vad.silero import SileroVADAnalyzer +from pipecat.transports.services.daily import DailyParams, DailyTransport, DailyTransportMessageFrame +from pipecat.services.openai import OpenAILLMService, OpenAILLMContext +from pipecat.services.deepgram import DeepgramSTTService +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.pipeline import Pipeline +from pipecat.processors.logger import FrameLogger +from pipecat.frames.frames import LLMMessagesFrame + + +from fastbothelpers import ( + GreedyLLMAggregator, + ClearableDeepgramTTSService, + VADGate, + AudioVolumeTimer, + TranscriptionTimingLogger +) + + +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, + transcription_enabled=False, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.650)), + vad_audio_passthrough=True + ) + ) + + stt = DeepgramSTTService( + api_key=os.getenv("DEEPGRAM_API_KEY"), + **({'url': url} if (url := os.getenv("DEEPGRAM_STT_URL")) else {}) + + ) + + tts = ClearableDeepgramTTSService( + name="STT", + aiohttp_session=session, + api_key=os.getenv("DEEPGRAM_API_KEY"), + voice="aura-asteria-en", + **({'base_url': url} if (url := os.getenv("DEEPGRAM_TTS_BASE_URL")) else {}) + ) + + llm = OpenAILLMService( + name="LLM", + # To use OpenAI + api_key=os.getenv("OPENAI_API_KEY"), + model=os.getenv("OPENAI_MODEL"), + base_url=os.getenv("OPENAI_BASE_URL") + ) + + messages = [ + { + "role": "system", + "content": """You are a helpful assistant in an audio conversation. + +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. Be concise in your answers to basic questions. If you are asked to elaborate or tell a story, provide a longer response. +""", + }, + ] + + ctx = OpenAILLMContext() + greedy = GreedyLLMAggregator(name="greedy", context=ctx) + gate = VADGate(name="gate", vad_analyzer=transport.input().vad_analyzer(), context=ctx) + avt = AudioVolumeTimer() + tl = TranscriptionTimingLogger(avt) + + pipeline = Pipeline([ + transport.input(), # Transport user input + avt, + stt, + tl, + greedy, + llm, # LLM + tts, # TTS + gate, + transport.output(), # Transport bot output + # FrameLogger() + ]) + + task = PipelineTask( + pipeline, + PipelineParams( + allow_interruptions=True, + enable_metrics=True, + report_only_initial_ttfb=True + )) + + # When a participant joins, start transcription for that participant so the + # bot can "hear" and respond to them. + # @ transport.event_handler("on_participant_joined") + # async def on_participant_joined(transport, participant): + # transport.capture_participant_transcription(participant["id"]) + + # When the first participant joins, the bot should introduce itself. + @ transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + messages.append( + {"role": "system", "content": "Please introduce yourself to the user."}) + await task.queue_frames([LLMMessagesFrame(messages)]) + + # Handle "latency-ping" messages. The client will send app messages that look like + # this: + # { "latency-ping": { ts: }} + # + # We want to send an immediate pong back to the client from this handler function. + # Also, we will push a frame into the top of the pipeline and send it after the + # + @ transport.event_handler("on_app_message") + async def on_app_message(transport, message, sender): + try: + if "latency-ping" in message: + logger.debug(f"Received latency ping app message: {message}") + ts = message["latency-ping"]["ts"] + # Send immediately + transport.output().send_message(DailyTransportMessageFrame( + message={"latency-pong-msg-handler": {"ts": ts}}, + participant_id=sender)) + # And push to the pipeline for the Daily transport.output to send + await tma_in.push_frame( + DailyTransportMessageFrame( + message={"latency-pong-pipeline-delivery": {"ts": ts}}, + participant_id=sender)) + except Exception as e: + logger.debug(f"message handling error: {e} - {message}") + + runner = PipelineRunner() + await runner.run(task) + + +if __name__ == "__main__": + (url, token) = configure() + asyncio.run(main(url, token))