From 87434460f5b24eac6e6a8deed398948f4b991b37 Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Sat, 2 Nov 2024 15:53:58 -0700 Subject: [PATCH 1/8] temp hacking --- .../22b-natural-conversation-anthropic.py | 212 ++++++++++++++++++ 1 file changed, 212 insertions(+) create mode 100644 examples/foundational/22b-natural-conversation-anthropic.py diff --git a/examples/foundational/22b-natural-conversation-anthropic.py b/examples/foundational/22b-natural-conversation-anthropic.py new file mode 100644 index 000000000..3e1d11c4b --- /dev/null +++ b/examples/foundational/22b-natural-conversation-anthropic.py @@ -0,0 +1,212 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import aiohttp +import os +import sys + +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.frames.frames import LLMMessagesFrame, TextFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.parallel_pipeline import ParallelPipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.gated_openai_llm_context import GatedOpenAILLMContextAggregator +from pipecat.processors.aggregators.openai_llm_context import ( + OpenAILLMContext, + OpenAILLMContextFrame, +) +from pipecat.processors.filters.null_filter import NullFilter +from pipecat.processors.filters.wake_notifier_filter import WakeNotifierFilter +from pipecat.processors.user_idle_processor import UserIdleProcessor +from pipecat.services.cartesia import CartesiaTTSService +from pipecat.services.deepgram import DeepgramSTTService +from pipecat.services.openai import OpenAILLMService +from pipecat.sync.event_notifier import EventNotifier +from pipecat.transports.services.daily import DailyParams, DailyTransport +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.frames.frames import Frame + + +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(): + async with aiohttp.ClientSession() as session: + (room_url, _) = await configure(session) + + transport = DailyTransport( + room_url, + None, + "Respond bot", + DailyParams( + audio_out_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + vad_audio_passthrough=True, + ), + ) + + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady + ) + + # This is the LLM that will be used to detect if the user has finished a + # statement. This doesn't really need to be an LLM, we could use NLP + # libraries for that, but it was easier as an example because we + # leverage the context aggregators. + statement_llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + + statement_messages = [ + { + "role": "system", + "content": "Determine if the user's statement is a complete sentence or question, ending in a natural pause or punctuation. Return 'YES' if it is complete and 'NO' if it seems to leave a thought unfinished.", + }, + ] + + statement_context = OpenAILLMContext(statement_messages) + statement_context_aggregator = statement_llm.create_context_aggregator(statement_context) + + # This is the regular LLM. + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + + messages = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", + }, + ] + + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) + + # We have instructed the LLM to return 'YES' if it thinks the user + # completed a sentence. So, if it's 'YES' we will return true in this + # predicate which will wake up the notifier. + async def wake_check_filter(frame): + logger.debug(f"Completeness check frame: {frame}") + return frame.text == "YES" + + # This is a notifier that we use to synchronize the two LLMs. + notifier = EventNotifier() + + # This a filter that will wake up the notifier if the given predicate + # (wake_check_filter) returns true. + completeness_check = WakeNotifierFilter( + notifier, types=(TextFrame,), filter=wake_check_filter + ) + + # This processor keeps the last context and will let it through once the + # notifier is woken up. + gated_context_aggregator = GatedOpenAILLMContextAggregator(notifier) + + # Notify if the user hasn't said anything. + async def user_idle_notifier(frame): + await notifier.notify() + + # Sometimes the LLM will fail detecting if a user has completed a + # sentence, this will wake up the notifier if that happens. + user_idle = UserIdleProcessor(callback=user_idle_notifier, timeout=3.0) + + class StatementJudgeContextFilter(FrameProcessor): + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, OpenAILLMContextFrame): + logger.debug(f"Context Frame: {frame}") + await self.push_frame(frame, direction) + + class GatedTTSOutput(FrameProcessor): + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + await self.push_frame(frame, direction) + + # The ParallePipeline input are the user transcripts. We have two + # contexts. The first one will be used to determine if the user finished + # a statement and if so the notifier will be woken up. The second + # context is simply the regular context but it's gated waiting for the + # notifier to be woken up. + pipeline = Pipeline( + [ + transport.input(), + stt, + ParallelPipeline( + [ + statement_context_aggregator.user(), + statement_llm, + completeness_check, + NullFilter(), + ], + [context_aggregator.user(), gated_context_aggregator, llm], + ), + user_idle, + tts, # TTS + transport.output(), + context_aggregator.assistant(), + ] + ) + + pipeline_x = Pipeline( + [ + transport.input(), + stt, + user_idle, + context_aggregator.user(), + ParallelPipeline( + [ + StatementJudgeContextFilter(), + statement_llm, + completeness_check, + NullFilter(), + ], + [ + llm, + tts, + GatedTTSOutput(), + ], + ), + transport.output(), + context_aggregator.assistant(), + ] + ) + + task = PipelineTask( + # pipeline, + pipeline_x, + PipelineParams( + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + report_only_initial_ttfb=True, + ), + ) + + @transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + await transport.capture_participant_transcription(participant["id"]) + # Kick off the conversation. + messages.append({"role": "system", "content": "Please introduce yourself to the user."}) + await task.queue_frames([LLMMessagesFrame(messages)]) + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) From 55a81df84fb3dcb477d0683866b4c064868222c4 Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Sun, 3 Nov 2024 21:40:41 -0800 Subject: [PATCH 2/8] contributing to llm-as-judge phrase endpointing work --- ...ersation-anthropic.py => 22b-natural-conversation-proposal.py} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename examples/foundational/{22b-natural-conversation-anthropic.py => 22b-natural-conversation-proposal.py} (100%) diff --git a/examples/foundational/22b-natural-conversation-anthropic.py b/examples/foundational/22b-natural-conversation-proposal.py similarity index 100% rename from examples/foundational/22b-natural-conversation-anthropic.py rename to examples/foundational/22b-natural-conversation-proposal.py From bd435d9e62d6cf66bcb32be7f7ac77ab7cf00c29 Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Mon, 4 Nov 2024 07:02:55 -0800 Subject: [PATCH 3/8] missing commit --- .../22b-natural-conversation-proposal.py | 277 +++++++++++++----- 1 file changed, 203 insertions(+), 74 deletions(-) diff --git a/examples/foundational/22b-natural-conversation-proposal.py b/examples/foundational/22b-natural-conversation-proposal.py index 3e1d11c4b..e0a7f71ba 100644 --- a/examples/foundational/22b-natural-conversation-proposal.py +++ b/examples/foundational/22b-natural-conversation-proposal.py @@ -4,10 +4,11 @@ # SPDX-License-Identifier: BSD 2-Clause License # -import asyncio import aiohttp +import asyncio import os import sys +import time from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.frames.frames import LLMMessagesFrame, TextFrame @@ -15,21 +16,29 @@ from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.parallel_pipeline import ParallelPipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask -from pipecat.processors.aggregators.gated_openai_llm_context import GatedOpenAILLMContextAggregator from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContext, - OpenAILLMContextFrame, ) -from pipecat.processors.filters.null_filter import NullFilter -from pipecat.processors.filters.wake_notifier_filter import WakeNotifierFilter -from pipecat.processors.user_idle_processor import UserIdleProcessor from pipecat.services.cartesia import CartesiaTTSService from pipecat.services.deepgram import DeepgramSTTService from pipecat.services.openai import OpenAILLMService from pipecat.sync.event_notifier import EventNotifier from pipecat.transports.services.daily import DailyParams, DailyTransport -from pipecat.processors.frame_processor import FrameDirection, FrameProcessor -from pipecat.frames.frames import Frame +from pipecat.processors.frame_processor import FrameProcessor, FrameDirection +from pipecat.frames.frames import ( + CancelFrame, + EndFrame, + Frame, + StartFrame, + StartInterruptionFrame, + SystemFrame, + TranscriptionFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, +) +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame +from pipecat.sync.base_notifier import BaseNotifier +from pipecat.processors.filters.function_filter import FunctionFilter from runner import configure @@ -44,6 +53,144 @@ logger.remove(0) logger.add(sys.stderr, level="DEBUG") +classifier_statement = "Determine if the user's statement ends with a complete sentence or question. The user text is transcribed speech. It may contain multiple fragments concatentated together. Categorize the text as either complete with the user now expecting a response, or incomplete. Return 'YES' if text is likely complete and the user is expecting a response. Return 'NO' if the text seems to be a partial expression or unfinished thought." + + +class StatementJudgeContextFilter(FrameProcessor): + 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) + return + + # We only want to handle OpenAILLMContextFrames, and only want to push a simple + # messages frame that contains a system prompt and the most recent user messages, + # concatenated. + if isinstance(frame, OpenAILLMContextFrame): + logger.debug(f"Context Frame: {frame}") + # Take text content from the most recent user messages. + messages = frame.context.messages + user_text_messages = [] + last_assistant_message = None + for message in reversed(messages): + if message["role"] != "user": + if message["role"] == "assistant": + last_assistant_message = message + break + if isinstance(message["content"], str): + user_text_messages.append(message["content"]) + elif isinstance(message["content"], list): + for content in message["content"]: + if content["type"] == "text": + user_text_messages.append(content["text"]) + # If we have any user text content, push an LLMMessagesFrame + if user_text_messages: + logger.debug(f"User text messages: {user_text_messages}") + user_message = " ".join(reversed(user_text_messages)) + logger.debug(f"User message: {user_message}") + messages = [ + { + "role": "system", + "content": classifier_statement, + } + ] + if last_assistant_message: + messages.append(last_assistant_message) + messages.append({"role": "user", "content": user_message}) + await self.push_frame(LLMMessagesFrame(messages)) + + +class CompletenessCheck(FrameProcessor): + def __init__(self, complete_notifier: BaseNotifier, incomplete_notifier: BaseNotifier): + super().__init__() + self._complete_notifier = complete_notifier + self._incomplete_notifier = incomplete_notifier + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, TextFrame) and frame.text == "YES": + logger.debug("Completeness check YES") + await self.push_frame(UserStoppedSpeakingFrame()) + await self._complete_notifier.notify() + elif isinstance(frame, TextFrame) and frame.text == "NO": + logger.debug("Completeness check NO") + await self._incomplete_notifier.notify() + + +class OutputGate(FrameProcessor): + def __init__( + self, complete_notifier: BaseNotifier, incomplete_notifier: BaseNotifier, **kwargs + ): + super().__init__(**kwargs) + self._gate_open = False + self._frames_buffer = [] + self._complete_notifier = complete_notifier + self._incomplete_notifier = incomplete_notifier + + def close_gate(self): + self._gate_open = False + + def open_gate(self): + self._gate_open = True + + 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): + if isinstance(frame, StartFrame): + await self._start() + if isinstance(frame, (EndFrame, CancelFrame)): + await self._stop() + if isinstance(frame, StartInterruptionFrame): + self._frames_buffer = [] + self.close_gate() + await self.push_frame(frame, direction) + return + + # Ignore frames that are not following the direction of this gate. + if direction != FrameDirection.DOWNSTREAM: + await self.push_frame(frame, direction) + return + + if self._gate_open: + await self.push_frame(frame, direction) + return + + self._frames_buffer.append((frame, direction)) + + async def _start(self): + self._frames_buffer = [] + self._gate_task = self.get_event_loop().create_task(self._gate_task_handler()) + self._interrupt_task = self.get_event_loop().create_task(self._interrupt_task_handler()) + + async def _stop(self): + self._gate_task.cancel() + await self._gate_task + + async def _gate_task_handler(self): + while True: + try: + await self._complete_notifier.wait() + self.open_gate() + for frame, direction in self._frames_buffer: + await self.push_frame(frame, direction) + self._frames_buffer = [] + except asyncio.CancelledError: + break + + async def _interrupt_task_handler(self): + while True: + try: + await self._incomplete_notifier.wait() + await self.push_frame(StartInterruptionFrame(), FrameDirection.UPSTREAM) + self._frames_buffer = [] + + except asyncio.CancelledError: + break + + async def main(): async with aiohttp.ClientSession() as session: (room_url, _) = await configure(session) @@ -69,20 +216,9 @@ async def main(): # This is the LLM that will be used to detect if the user has finished a # statement. This doesn't really need to be an LLM, we could use NLP - # libraries for that, but it was easier as an example because we - # leverage the context aggregators. + # libraries for that, but we have the machinery to use an LLM, so we might as well! statement_llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") - statement_messages = [ - { - "role": "system", - "content": "Determine if the user's statement is a complete sentence or question, ending in a natural pause or punctuation. Return 'YES' if it is complete and 'NO' if it seems to leave a thought unfinished.", - }, - ] - - statement_context = OpenAILLMContext(statement_messages) - statement_context_aggregator = statement_llm.create_context_aggregator(statement_context) - # This is the regular LLM. llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") @@ -105,79 +241,58 @@ async def main(): # This is a notifier that we use to synchronize the two LLMs. notifier = EventNotifier() + # rename/comment? + interrupt_notifier = EventNotifier() - # This a filter that will wake up the notifier if the given predicate - # (wake_check_filter) returns true. - completeness_check = WakeNotifierFilter( - notifier, types=(TextFrame,), filter=wake_check_filter + # This sends a UserStoppedSpeakingFrame and triggers the notifier event + completeness_check = CompletenessCheck( + complete_notifier=notifier, incomplete_notifier=interrupt_notifier ) - # This processor keeps the last context and will let it through once the - # notifier is woken up. - gated_context_aggregator = GatedOpenAILLMContextAggregator(notifier) + # # Notify if the user hasn't said anything. + # async def user_idle_notifier(frame): + # await notifier.notify() - # Notify if the user hasn't said anything. - async def user_idle_notifier(frame): - await notifier.notify() + # # Sometimes the LLM will fail detecting if a user has completed a + # # sentence, this will wake up the notifier if that happens. + # user_idle = UserIdleProcessor(callback=user_idle_notifier, timeout=10.0) - # Sometimes the LLM will fail detecting if a user has completed a - # sentence, this will wake up the notifier if that happens. - user_idle = UserIdleProcessor(callback=user_idle_notifier, timeout=3.0) + bot_output_gate = OutputGate( + complete_notifier=notifier, incomplete_notifier=interrupt_notifier + ) - class StatementJudgeContextFilter(FrameProcessor): - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - if isinstance(frame, OpenAILLMContextFrame): - logger.debug(f"Context Frame: {frame}") - await self.push_frame(frame, direction) + async def block_user_stopped_speaking(frame): + return not isinstance(frame, UserStoppedSpeakingFrame) - class GatedTTSOutput(FrameProcessor): - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - await self.push_frame(frame, direction) + async def pass_only_llm_trigger_frames(frame): + return isinstance(frame, OpenAILLMContextFrame) or isinstance(frame, LLMMessagesFrame) - # The ParallePipeline input are the user transcripts. We have two - # contexts. The first one will be used to determine if the user finished - # a statement and if so the notifier will be woken up. The second - # context is simply the regular context but it's gated waiting for the - # notifier to be woken up. pipeline = Pipeline( [ transport.input(), stt, - ParallelPipeline( - [ - statement_context_aggregator.user(), - statement_llm, - completeness_check, - NullFilter(), - ], - [context_aggregator.user(), gated_context_aggregator, llm], - ), - user_idle, - tts, # TTS - transport.output(), - context_aggregator.assistant(), - ] - ) - - pipeline_x = Pipeline( - [ - transport.input(), - stt, - user_idle, + # user_idle, context_aggregator.user(), ParallelPipeline( [ + # Pass everything except UserStoppedSpeaking to the elements after + # this ParallelPipeline + FunctionFilter(filter=block_user_stopped_speaking), + ], + [ + # Ignore everything except an OpenAILLMContextFrame. Pass a specially constructed + # LLMMessagesFrame to the statement classifier LLM. The only frame this + # sub-pipeline will output is a UserStoppedSpeakingFrame. StatementJudgeContextFilter(), statement_llm, completeness_check, - NullFilter(), ], [ + # Block everything except OpenAILLMContextFrame and LLMMessagesFrame + FunctionFilter(filter=pass_only_llm_trigger_frames), llm, tts, - GatedTTSOutput(), + bot_output_gate, # Buffer all llm/tts output until notified. ], ), transport.output(), @@ -186,8 +301,7 @@ async def main(): ) task = PipelineTask( - # pipeline, - pipeline_x, + pipeline, PipelineParams( allow_interruptions=True, enable_metrics=True, @@ -203,8 +317,23 @@ async def main(): messages.append({"role": "system", "content": "Please introduce yourself to the user."}) await task.queue_frames([LLMMessagesFrame(messages)]) - runner = PipelineRunner() + @transport.event_handler("on_app_message") + async def on_app_message(transport, message, sender): + logger.debug(f"Received app message: {message} - {sender}") + if "message" not in message: + return + await task.queue_frames( + [ + UserStartedSpeakingFrame(), + TranscriptionFrame( + user_id=sender, timestamp=time.time(), text=message["message"] + ), + UserStoppedSpeakingFrame(), + ] + ) + + runner = PipelineRunner() await runner.run(task) From b56c789ae4d60c944874c2a216d6fb488d78356f Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Mon, 4 Nov 2024 13:28:35 -0800 Subject: [PATCH 4/8] fixes for proposed judge pipeline --- .../22b-natural-conversation-proposal.py | 80 +++++++++---------- .../processors/filters/function_filter.py | 16 +++- 2 files changed, 51 insertions(+), 45 deletions(-) diff --git a/examples/foundational/22b-natural-conversation-proposal.py b/examples/foundational/22b-natural-conversation-proposal.py index e0a7f71ba..cae117df3 100644 --- a/examples/foundational/22b-natural-conversation-proposal.py +++ b/examples/foundational/22b-natural-conversation-proposal.py @@ -31,6 +31,7 @@ from pipecat.frames.frames import ( Frame, StartFrame, StartInterruptionFrame, + StopInterruptionFrame, SystemFrame, TranscriptionFrame, UserStartedSpeakingFrame, @@ -39,6 +40,7 @@ from pipecat.frames.frames import ( from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame from pipecat.sync.base_notifier import BaseNotifier from pipecat.processors.filters.function_filter import FunctionFilter +from pipecat.processors.user_idle_processor import UserIdleProcessor from runner import configure @@ -53,10 +55,14 @@ logger.remove(0) logger.add(sys.stderr, level="DEBUG") -classifier_statement = "Determine if the user's statement ends with a complete sentence or question. The user text is transcribed speech. It may contain multiple fragments concatentated together. Categorize the text as either complete with the user now expecting a response, or incomplete. Return 'YES' if text is likely complete and the user is expecting a response. Return 'NO' if the text seems to be a partial expression or unfinished thought." +classifier_statement = "Determine if the user's statement ends with a complete thought and you should respond. The user text is transcribed speech. It may contain multiple fragments concatentated together. You are trying to determine only the completeness of the last user statement. The previous assistant statement is provided only for context. Categorize the text as either complete with the user now expecting a response, or incomplete. Return 'YES' if text is likely complete and the user is expecting a response. Return 'NO' if the text seems to be a partial expression or unfinished thought." class StatementJudgeContextFilter(FrameProcessor): + def __init__(self, notifier: BaseNotifier, **kwargs): + super().__init__(**kwargs) + self._notifier = notifier + async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) # We must not block system frames. @@ -64,7 +70,12 @@ class StatementJudgeContextFilter(FrameProcessor): await self.push_frame(frame, direction) return - # We only want to handle OpenAILLMContextFrames, and only want to push a simple + # Just treat an LLMMessagesFrame as complete, no matter what. + if isinstance(frame, LLMMessagesFrame): + await self._notifier.notify() + return + + # Otherwise, we only want to handle OpenAILLMContextFrames, and only want to push a simple # messages frame that contains a system prompt and the most recent user messages, # concatenated. if isinstance(frame, OpenAILLMContextFrame): @@ -102,31 +113,26 @@ class StatementJudgeContextFilter(FrameProcessor): class CompletenessCheck(FrameProcessor): - def __init__(self, complete_notifier: BaseNotifier, incomplete_notifier: BaseNotifier): + def __init__(self, notifier: BaseNotifier): super().__init__() - self._complete_notifier = complete_notifier - self._incomplete_notifier = incomplete_notifier + self._notifier = notifier async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) if isinstance(frame, TextFrame) and frame.text == "YES": logger.debug("Completeness check YES") await self.push_frame(UserStoppedSpeakingFrame()) - await self._complete_notifier.notify() + await self._notifier.notify() elif isinstance(frame, TextFrame) and frame.text == "NO": logger.debug("Completeness check NO") - await self._incomplete_notifier.notify() class OutputGate(FrameProcessor): - def __init__( - self, complete_notifier: BaseNotifier, incomplete_notifier: BaseNotifier, **kwargs - ): + def __init__(self, notifier: BaseNotifier, **kwargs): super().__init__(**kwargs) self._gate_open = False self._frames_buffer = [] - self._complete_notifier = complete_notifier - self._incomplete_notifier = incomplete_notifier + self._notifier = notifier def close_gate(self): self._gate_open = False @@ -163,7 +169,6 @@ class OutputGate(FrameProcessor): async def _start(self): self._frames_buffer = [] self._gate_task = self.get_event_loop().create_task(self._gate_task_handler()) - self._interrupt_task = self.get_event_loop().create_task(self._interrupt_task_handler()) async def _stop(self): self._gate_task.cancel() @@ -172,7 +177,7 @@ class OutputGate(FrameProcessor): async def _gate_task_handler(self): while True: try: - await self._complete_notifier.wait() + await self._notifier.wait() self.open_gate() for frame, direction in self._frames_buffer: await self.push_frame(frame, direction) @@ -180,16 +185,6 @@ class OutputGate(FrameProcessor): except asyncio.CancelledError: break - async def _interrupt_task_handler(self): - while True: - try: - await self._incomplete_notifier.wait() - await self.push_frame(StartInterruptionFrame(), FrameDirection.UPSTREAM) - self._frames_buffer = [] - - except asyncio.CancelledError: - break - async def main(): async with aiohttp.ClientSession() as session: @@ -241,37 +236,39 @@ async def main(): # This is a notifier that we use to synchronize the two LLMs. notifier = EventNotifier() - # rename/comment? - interrupt_notifier = EventNotifier() + + # This turns the LLM context into an inference request to classify the user's speech + # as complete or incomplete. + statement_judge_context_filter = StatementJudgeContextFilter(notifier=notifier) # This sends a UserStoppedSpeakingFrame and triggers the notifier event - completeness_check = CompletenessCheck( - complete_notifier=notifier, incomplete_notifier=interrupt_notifier - ) + completeness_check = CompletenessCheck(notifier=notifier) # # Notify if the user hasn't said anything. - # async def user_idle_notifier(frame): - # await notifier.notify() + async def user_idle_notifier(frame): + await notifier.notify() - # # Sometimes the LLM will fail detecting if a user has completed a - # # sentence, this will wake up the notifier if that happens. - # user_idle = UserIdleProcessor(callback=user_idle_notifier, timeout=10.0) + # Sometimes the LLM will fail detecting if a user has completed a + # sentence, this will wake up the notifier if that happens. + user_idle = UserIdleProcessor(callback=user_idle_notifier, timeout=5.0) - bot_output_gate = OutputGate( - complete_notifier=notifier, incomplete_notifier=interrupt_notifier - ) + bot_output_gate = OutputGate(notifier=notifier) async def block_user_stopped_speaking(frame): return not isinstance(frame, UserStoppedSpeakingFrame) async def pass_only_llm_trigger_frames(frame): - return isinstance(frame, OpenAILLMContextFrame) or isinstance(frame, LLMMessagesFrame) + return ( + isinstance(frame, OpenAILLMContextFrame) + or isinstance(frame, LLMMessagesFrame) + or isinstance(frame, StartInterruptionFrame) + or isinstance(frame, StopInterruptionFrame) + ) pipeline = Pipeline( [ transport.input(), stt, - # user_idle, context_aggregator.user(), ParallelPipeline( [ @@ -283,7 +280,7 @@ async def main(): # Ignore everything except an OpenAILLMContextFrame. Pass a specially constructed # LLMMessagesFrame to the statement classifier LLM. The only frame this # sub-pipeline will output is a UserStoppedSpeakingFrame. - StatementJudgeContextFilter(), + statement_judge_context_filter, statement_llm, completeness_check, ], @@ -291,10 +288,11 @@ async def main(): # Block everything except OpenAILLMContextFrame and LLMMessagesFrame FunctionFilter(filter=pass_only_llm_trigger_frames), llm, - tts, bot_output_gate, # Buffer all llm/tts output until notified. ], ), + tts, + user_idle, transport.output(), context_aggregator.assistant(), ] diff --git a/src/pipecat/processors/filters/function_filter.py b/src/pipecat/processors/filters/function_filter.py index ba1f706a7..e38cea3e0 100644 --- a/src/pipecat/processors/filters/function_filter.py +++ b/src/pipecat/processors/filters/function_filter.py @@ -11,19 +11,27 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor class FunctionFilter(FrameProcessor): - def __init__(self, filter: Callable[[Frame], Awaitable[bool]]): + def __init__( + self, + filter: Callable[[Frame], Awaitable[bool]], + direction: FrameDirection = FrameDirection.DOWNSTREAM, + ): super().__init__() self._filter = filter + self._direction = direction # # Frame processor # - def _should_passthrough_frame(self, frame): - return isinstance(frame, SystemFrame) + # Ignore system frames and frames that are not following the direction of this gate + def _should_passthrough_frame(self, frame, direction): + return isinstance(frame, SystemFrame) or direction != self._direction async def process_frame(self, frame: Frame, direction: FrameDirection): - passthrough = self._should_passthrough_frame(frame) + await super().process_frame(frame, direction) + + passthrough = self._should_passthrough_frame(frame, direction) allowed = await self._filter(frame) if passthrough or allowed: await self.push_frame(frame, direction) From b6c2c1f73044c9b03f915af2b8319bb4a0be20f1 Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Mon, 4 Nov 2024 15:10:24 -0800 Subject: [PATCH 5/8] anthropic natural conversation example using claude haiku --- .../22c-natural-conversation-anthropic.py | 383 ++++++++++++++++++ 1 file changed, 383 insertions(+) create mode 100644 examples/foundational/22c-natural-conversation-anthropic.py diff --git a/examples/foundational/22c-natural-conversation-anthropic.py b/examples/foundational/22c-natural-conversation-anthropic.py new file mode 100644 index 000000000..9d56b2a5e --- /dev/null +++ b/examples/foundational/22c-natural-conversation-anthropic.py @@ -0,0 +1,383 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import aiohttp +import asyncio +import os +import sys +import time + +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.frames.frames import LLMMessagesFrame, TextFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.parallel_pipeline import ParallelPipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.openai_llm_context import ( + OpenAILLMContext, +) +from pipecat.services.cartesia import CartesiaTTSService +from pipecat.services.deepgram import DeepgramSTTService +from pipecat.services.anthropic import AnthropicLLMService +from pipecat.sync.event_notifier import EventNotifier +from pipecat.transports.services.daily import DailyParams, DailyTransport +from pipecat.processors.frame_processor import FrameProcessor, FrameDirection +from pipecat.frames.frames import ( + BotSpeakingFrame, + CancelFrame, + EndFrame, + Frame, + StartFrame, + StartInterruptionFrame, + StopInterruptionFrame, + SystemFrame, + TranscriptionFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, +) +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame +from pipecat.sync.base_notifier import BaseNotifier +from pipecat.processors.filters.function_filter import FunctionFilter +from pipecat.processors.user_idle_processor import UserIdleProcessor + + +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") + + +classifier_statement = """Determine if the user's statement ends with a complete thought and you should respond. + +The user text is transcribed speech. It may contain multiple fragments concatentated together. + +You are trying to determine only the completeness of the last user statement. The previous assistant statement is provided only for context. Categorize the text as either complete with the user now expecting a response, or incomplete. + +Return 'YES' if text is likely complete and the user is expecting a response. Return 'NO' if the text seems to be a partial expression or unfinished thought. + +Respond only with either YES or NO + +Examples: + +User: What's the capital of +Assistant: NO + +User: What's the captial of France? +Assistant: YES + +User: Tell me a story about +Assistant: NO + +User: Tell me a story about a dragon +Assistant YES + +User: Is there a +Assistant: NO + +User: Is there a large +Assistant: NO + +User: Is there a large lake near Chicago? +Assistant: YES + +User: When is the longest day of the year? +Assistant: YES +""" + + +class StatementJudgeContextFilter(FrameProcessor): + def __init__(self, notifier: BaseNotifier, **kwargs): + super().__init__(**kwargs) + self._notifier = notifier + + 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) + return + + # Just treat an LLMMessagesFrame as complete, no matter what. + if isinstance(frame, LLMMessagesFrame): + await self._notifier.notify() + return + + # Otherwise, we only want to handle OpenAILLMContextFrames, and only want to push a simple + # messages frame that contains a system prompt and the most recent user messages, + # concatenated. + if isinstance(frame, OpenAILLMContextFrame): + logger.debug(f"Context Frame: {frame}") + # Take text content from the most recent user messages. + messages = frame.context.messages + user_text_messages = [] + last_assistant_message = None + for message in reversed(messages): + if message["role"] != "user": + if message["role"] == "assistant": + last_assistant_message = message + break + if isinstance(message["content"], str): + user_text_messages.append(message["content"]) + elif isinstance(message["content"], list): + for content in message["content"]: + if content["type"] == "text": + user_text_messages.insert(0, content["text"]) + # If we have any user text content, push an LLMMessagesFrame + if user_text_messages: + logger.debug(f"User text messages: {user_text_messages}") + user_message = " ".join(reversed(user_text_messages)) + logger.debug(f"User message: {user_message}") + messages = [ + { + "role": "system", + "content": classifier_statement, + } + ] + if last_assistant_message: + messages.append(last_assistant_message) + messages.append({"role": "user", "content": user_message}) + await self.push_frame(LLMMessagesFrame(messages)) + + +class CompletenessCheck(FrameProcessor): + def __init__(self, notifier: BaseNotifier): + super().__init__() + self._notifier = notifier + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if not isinstance(frame, BotSpeakingFrame): + logger.debug(f"!!! RESULT Completeness check frame: {frame}") + + if isinstance(frame, TextFrame) and frame.text == "YES": + logger.debug("Completeness check YES") + await self.push_frame(UserStoppedSpeakingFrame()) + await self._notifier.notify() + elif isinstance(frame, TextFrame) and frame.text == "NO": + logger.debug("Completeness check NO") + + +class OutputGate(FrameProcessor): + def __init__(self, notifier: BaseNotifier, **kwargs): + super().__init__(**kwargs) + self._gate_open = False + self._frames_buffer = [] + self._notifier = notifier + + def close_gate(self): + self._gate_open = False + + def open_gate(self): + self._gate_open = True + + 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): + if isinstance(frame, StartFrame): + await self._start() + if isinstance(frame, (EndFrame, CancelFrame)): + await self._stop() + if isinstance(frame, StartInterruptionFrame): + self._frames_buffer = [] + self.close_gate() + await self.push_frame(frame, direction) + return + + # Ignore frames that are not following the direction of this gate. + if direction != FrameDirection.DOWNSTREAM: + await self.push_frame(frame, direction) + return + + if self._gate_open: + await self.push_frame(frame, direction) + return + + self._frames_buffer.append((frame, direction)) + + async def _start(self): + self._frames_buffer = [] + self._gate_task = self.get_event_loop().create_task(self._gate_task_handler()) + + async def _stop(self): + self._gate_task.cancel() + await self._gate_task + + async def _gate_task_handler(self): + while True: + try: + await self._notifier.wait() + self.open_gate() + for frame, direction in self._frames_buffer: + await self.push_frame(frame, direction) + self._frames_buffer = [] + except asyncio.CancelledError: + break + + +async def main(): + async with aiohttp.ClientSession() as session: + (room_url, _) = await configure(session) + + transport = DailyTransport( + room_url, + None, + "Respond bot", + DailyParams( + audio_out_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + vad_audio_passthrough=True, + ), + ) + + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady + ) + + # This is the LLM that will be used to detect if the user has finished a + # statement. This doesn't really need to be an LLM, we could use NLP + # libraries for that, but we have the machinery to use an LLM, so we might as well! + statement_llm = AnthropicLLMService( + api_key=os.getenv("ANTHROPIC_API_KEY"), model="claude-3-5-haiku-20241022" + ) + + # This is the regular LLM. + llm = AnthropicLLMService( + api_key=os.getenv("ANTHROPIC_API_KEY"), model="claude-3-5-sonnet-20241022" + ) + + messages = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", + }, + ] + + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) + + # We have instructed the LLM to return 'YES' if it thinks the user + # completed a sentence. So, if it's 'YES' we will return true in this + # predicate which will wake up the notifier. + async def wake_check_filter(frame): + logger.debug(f"Completeness check frame: {frame}") + return frame.text == "YES" + + # This is a notifier that we use to synchronize the two LLMs. + notifier = EventNotifier() + + # This turns the LLM context into an inference request to classify the user's speech + # as complete or incomplete. + statement_judge_context_filter = StatementJudgeContextFilter(notifier=notifier) + + # This sends a UserStoppedSpeakingFrame and triggers the notifier event + completeness_check = CompletenessCheck(notifier=notifier) + + # # Notify if the user hasn't said anything. + async def user_idle_notifier(frame): + await notifier.notify() + + # Sometimes the LLM will fail detecting if a user has completed a + # sentence, this will wake up the notifier if that happens. + user_idle = UserIdleProcessor(callback=user_idle_notifier, timeout=5.0) + + bot_output_gate = OutputGate(notifier=notifier) + + async def block_user_stopped_speaking(frame): + return not isinstance(frame, UserStoppedSpeakingFrame) + + async def pass_only_llm_trigger_frames(frame): + return ( + isinstance(frame, OpenAILLMContextFrame) + or isinstance(frame, LLMMessagesFrame) + or isinstance(frame, StartInterruptionFrame) + or isinstance(frame, StopInterruptionFrame) + ) + + pipeline = Pipeline( + [ + transport.input(), + stt, + context_aggregator.user(), + ParallelPipeline( + [ + # Pass everything except UserStoppedSpeaking to the elements after + # this ParallelPipeline + FunctionFilter(filter=block_user_stopped_speaking), + ], + [ + # Ignore everything except an OpenAILLMContextFrame. Pass a specially constructed + # LLMMessagesFrame to the statement classifier LLM. The only frame this + # sub-pipeline will output is a UserStoppedSpeakingFrame. + statement_judge_context_filter, + statement_llm, + completeness_check, + ], + [ + # Block everything except OpenAILLMContextFrame and LLMMessagesFrame + FunctionFilter(filter=pass_only_llm_trigger_frames), + llm, + bot_output_gate, # Buffer all llm/tts output until notified. + ], + ), + tts, + user_idle, + transport.output(), + context_aggregator.assistant(), + ] + ) + + task = PipelineTask( + pipeline, + PipelineParams( + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + report_only_initial_ttfb=True, + ), + ) + + @transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + await transport.capture_participant_transcription(participant["id"]) + # Kick off the conversation. + messages.append({"role": "user", "content": "Please introduce yourself to the user."}) + await task.queue_frames([LLMMessagesFrame(messages)]) + + @transport.event_handler("on_app_message") + async def on_app_message(transport, message, sender): + logger.debug(f"Received app message: {message} - {sender}") + if "message" not in message: + return + + await task.queue_frames( + [ + UserStartedSpeakingFrame(), + TranscriptionFrame( + user_id=sender, timestamp=time.time(), text=message["message"] + ), + UserStoppedSpeakingFrame(), + ] + ) + + runner = PipelineRunner() + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) From 91ac40307ed134e44664507c151d70489c2371d4 Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Mon, 4 Nov 2024 16:28:31 -0800 Subject: [PATCH 6/8] small fix and more prompt examples --- .../22b-natural-conversation-proposal.py | 2 +- .../22c-natural-conversation-anthropic.py | 86 +++++++++++++++---- 2 files changed, 69 insertions(+), 19 deletions(-) diff --git a/examples/foundational/22b-natural-conversation-proposal.py b/examples/foundational/22b-natural-conversation-proposal.py index cae117df3..2deeb3da4 100644 --- a/examples/foundational/22b-natural-conversation-proposal.py +++ b/examples/foundational/22b-natural-conversation-proposal.py @@ -94,7 +94,7 @@ class StatementJudgeContextFilter(FrameProcessor): elif isinstance(message["content"], list): for content in message["content"]: if content["type"] == "text": - user_text_messages.append(content["text"]) + user_text_messages.insert(0, content["text"]) # If we have any user text content, push an LLMMessagesFrame if user_text_messages: logger.debug(f"User text messages: {user_text_messages}") diff --git a/examples/foundational/22c-natural-conversation-anthropic.py b/examples/foundational/22c-natural-conversation-anthropic.py index 9d56b2a5e..ebc921628 100644 --- a/examples/foundational/22c-natural-conversation-anthropic.py +++ b/examples/foundational/22c-natural-conversation-anthropic.py @@ -26,7 +26,6 @@ from pipecat.sync.event_notifier import EventNotifier from pipecat.transports.services.daily import DailyParams, DailyTransport from pipecat.processors.frame_processor import FrameProcessor, FrameDirection from pipecat.frames.frames import ( - BotSpeakingFrame, CancelFrame, EndFrame, Frame, @@ -58,13 +57,22 @@ logger.add(sys.stderr, level="DEBUG") classifier_statement = """Determine if the user's statement ends with a complete thought and you should respond. -The user text is transcribed speech. It may contain multiple fragments concatentated together. +The user text is transcribed speech. You are trying to determine if: -You are trying to determine only the completeness of the last user statement. The previous assistant statement is provided only for context. Categorize the text as either complete with the user now expecting a response, or incomplete. +1. the user has finished talking and expects a response from you, or +2. this statement is incomplete and the user will continue talking + +A previous assistant response is provided for additional context. But you are only evaluating the user text. + +The user text may contain multiple fragments concatentated together. There may be repeated words or mistakes in the transcription. There may be grammatical errors. There may be extra punctuation. Ignore all of that. Interpret the transcribed text as text that would have been spoken. Then consider only whether the user has finished speaking and is expecting a response. + +Categorize the last user statement as either complete with the user now expecting a response, or incomplete. Return 'YES' if text is likely complete and the user is expecting a response. Return 'NO' if the text seems to be a partial expression or unfinished thought. -Respond only with either YES or NO +If you are not sure, respond with your best guess. If the user is expecting a response, respond with YES. If the user is not expecting a response, respond with NO. Always output either YES or NO and no other text. + +Respond only YES or NO Examples: @@ -91,6 +99,47 @@ Assistant: YES User: When is the longest day of the year? Assistant: YES + +User: When when is the longest day of the year +Assistant: YES + +User: When when is the +ASSISTANT: NO + +User: What is the um I u +Assistant: NO + +User: What is the um i u largest city in the world +Assistant: YES + +User: How much does a how much does an adult elephant weigh? +Assistant: YES + +User: How much does a how much does +Assistant: NO + +User: What can you tell me All the +Assistant: NO + +User: What can you tell me All the prime numbers less than 100 +Assistant: YES + +User: What's the what's the length of the Amazon River? +Assistant: YES + +User: What's what's the length of the Amazon River? +Assistant: YES + +User: What's what's the length of the Amazon River +Assistant: YES + +User: What's what's the best way to get a coffee stain out of a white shirt +Assistant: YES +""" + +conversational_system_message = """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. + +Please be very concise in your responses. Unless you are explicitly asked to do otherwise, give me the shortest complete answer possible without unnecessary elaboration. Generally you should answer with a single sentence. """ @@ -115,7 +164,6 @@ class StatementJudgeContextFilter(FrameProcessor): # messages frame that contains a system prompt and the most recent user messages, # concatenated. if isinstance(frame, OpenAILLMContextFrame): - logger.debug(f"Context Frame: {frame}") # Take text content from the most recent user messages. messages = frame.context.messages user_text_messages = [] @@ -133,9 +181,8 @@ class StatementJudgeContextFilter(FrameProcessor): user_text_messages.insert(0, content["text"]) # If we have any user text content, push an LLMMessagesFrame if user_text_messages: - logger.debug(f"User text messages: {user_text_messages}") user_message = " ".join(reversed(user_text_messages)) - logger.debug(f"User message: {user_message}") + logger.debug(f"!!! {user_message}") messages = [ { "role": "system", @@ -156,15 +203,12 @@ class CompletenessCheck(FrameProcessor): async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) - if not isinstance(frame, BotSpeakingFrame): - logger.debug(f"!!! RESULT Completeness check frame: {frame}") - if isinstance(frame, TextFrame) and frame.text == "YES": - logger.debug("Completeness check YES") + logger.debug("!!! Completeness check YES") await self.push_frame(UserStoppedSpeakingFrame()) await self._notifier.notify() elif isinstance(frame, TextFrame) and frame.text == "NO": - logger.debug("Completeness check NO") + logger.debug("!!! Completeness check NO") class OutputGate(FrameProcessor): @@ -253,18 +297,21 @@ async def main(): # statement. This doesn't really need to be an LLM, we could use NLP # libraries for that, but we have the machinery to use an LLM, so we might as well! statement_llm = AnthropicLLMService( - api_key=os.getenv("ANTHROPIC_API_KEY"), model="claude-3-5-haiku-20241022" + api_key=os.getenv("ANTHROPIC_API_KEY"), model="claude-3-5-haiku-20241022", name="Haiku" ) # This is the regular LLM. llm = AnthropicLLMService( - api_key=os.getenv("ANTHROPIC_API_KEY"), model="claude-3-5-sonnet-20241022" + api_key=os.getenv("ANTHROPIC_API_KEY"), + model="claude-3-5-sonnet-20241022", + name="Sonnet", + params=AnthropicLLMService.InputParams(enable_prompt_caching_beta=True), ) 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.", + "content": conversational_system_message, }, ] @@ -275,7 +322,6 @@ async def main(): # completed a sentence. So, if it's 'YES' we will return true in this # predicate which will wake up the notifier. async def wake_check_filter(frame): - logger.debug(f"Completeness check frame: {frame}") return frame.text == "YES" # This is a notifier that we use to synchronize the two LLMs. @@ -348,7 +394,6 @@ async def main(): allow_interruptions=True, enable_metrics=True, enable_usage_metrics=True, - report_only_initial_ttfb=True, ), ) @@ -356,7 +401,12 @@ async def main(): async def on_first_participant_joined(transport, participant): await transport.capture_participant_transcription(participant["id"]) # Kick off the conversation. - messages.append({"role": "user", "content": "Please introduce yourself to the user."}) + messages.append( + { + "role": "user", + "content": "Start by just saying \"Hello I'm ready.\" Don't say anything else.", + } + ) await task.queue_frames([LLMMessagesFrame(messages)]) @transport.event_handler("on_app_message") From ee53535f41485bc9399e21eaefe5f77cd3969f9a Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Fri, 8 Nov 2024 08:28:54 -0800 Subject: [PATCH 7/8] gemini audio-in with no transcription --- .../07p-interruptible-google-audio-in.py | 276 ++++++++++++++++++ .../aggregators/openai_llm_context.py | 28 ++ src/pipecat/services/google.py | 142 +++++---- 3 files changed, 387 insertions(+), 59 deletions(-) create mode 100644 examples/foundational/07p-interruptible-google-audio-in.py diff --git a/examples/foundational/07p-interruptible-google-audio-in.py b/examples/foundational/07p-interruptible-google-audio-in.py new file mode 100644 index 000000000..33bb30187 --- /dev/null +++ b/examples/foundational/07p-interruptible-google-audio-in.py @@ -0,0 +1,276 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import aiohttp +import asyncio +import os +import sys + +import google.ai.generativelanguage as glm + +from dataclasses import dataclass +from dotenv import load_dotenv +from loguru import logger +from runner import configure + +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.services.cartesia import CartesiaTTSService +from pipecat.services.google import GoogleLLMService +from pipecat.processors.frame_processor import FrameProcessor +from pipecat.transports.services.daily import DailyParams, DailyTransport +from pipecat.frames.frames import ( + LLMFullResponseStartFrame, + LLMFullResponseEndFrame, + InputAudioRawFrame, + Frame, + StartInterruptionFrame, + TextFrame, + TranscriptionFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, +) + +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + +marker = "|----|" +system_message = f""" +You are a helpful LLM in a WebRTC call. Your goals are to be helpful and brief in your responses. + +You are expert at transcribing audio to text. You will receive a mixture of audio and text input. When +asked to transcribe what the user said, output an exact, word-for-word transcription. + +Your output will be converted to audio so don't include special characters in your answers. + +Each time you answer, you should respond in three parts. + +1. Transcribe exactly what the user said. +2. Output the separator field '{marker}'. +3. Respond to the user's input in a succinct, helpful, creative way using only simple text and punctuation. + +Example: + +User: How many ounces are in a pound? + +You: How many ounces are in a pound? +{marker} +There are 16 ounces in a pound. +""" + + +@dataclass +class MagicDemoTranscriptionFrame(Frame): + text: str + + +class UserAudioCollector(FrameProcessor): + def __init__(self, context, user_context_aggregator): + super().__init__() + self._context = context + self._user_context_aggregator = user_context_aggregator + self._audio_frames = [] + self._start_secs = ( + 0.2 # this should match VAD_START_SECS but we'll just hardcode it for now + ) + self._user_speaking = False + + async def process_frame(self, frame, direction): + await super().process_frame(frame, direction) + + if isinstance(frame, TranscriptionFrame): + # We could gracefully handle both audio input and text/transcription input ... + # but let's leave that as an exercise to the reader. :-) + return + if isinstance(frame, UserStartedSpeakingFrame): + self._user_speaking = True + elif isinstance(frame, UserStoppedSpeakingFrame): + self._user_speaking = False + self._context.add_audio_frames_message(audio_frames=self._audio_frames) + await self._user_context_aggregator.push_frame( + self._user_context_aggregator.get_context_frame() + ) + elif isinstance(frame, InputAudioRawFrame): + if self._user_speaking: + self._audio_frames.append(frame) + else: + # Append the audio frame to our buffer. Treat the buffer as a ring buffer, dropping the oldest + # frames as necessary. Assume all audio frames have the same duration. + self._audio_frames.append(frame) + frame_duration = len(frame.audio) / 16 * frame.num_channels / frame.sample_rate + buffer_duration = frame_duration * len(self._audio_frames) + while buffer_duration > self._start_secs: + self._audio_frames.pop(0) + buffer_duration -= frame_duration + + await self.push_frame(frame, direction) + + +class TranscriptExtractor(FrameProcessor): + def __init__(self, context): + super().__init__() + self._context = context + self._accumulator = "" + self._processing_llm_response = False + self._accumulating_transcript = False + + def reset(self): + self._accumulator = "" + self._processing_llm_response = False + self._accumulating_transcript = False + + async def process_frame(self, frame, direction): + await super().process_frame(frame, direction) + if isinstance(frame, LLMFullResponseStartFrame): + self._processing_llm_response = True + self._accumulating_transcript = True + elif isinstance(frame, TextFrame) and self._processing_llm_response: + if self._accumulating_transcript: + text = frame.text + split_index = text.find(marker) + if split_index < 0: + self._accumulator += frame.text + # do not push this frame + return + else: + self._accumulating_transcript = False + self._accumulator += text[:split_index] + frame.text = text[split_index + len(marker) :] + await self.push_frame(frame) + return + elif isinstance(frame, LLMFullResponseEndFrame): + await self.push_frame(MagicDemoTranscriptionFrame(text=self._accumulator.strip())) + self.reset() + + await self.push_frame(frame, direction) + + +class TanscriptionContextFixup(FrameProcessor): + def __init__(self, context): + super().__init__() + self._context = context + self._transcript = "THIS IS A TRANSCRIPT" + + def swap_user_audio(self): + if not self._transcript: + return + message = self._context.messages[-2] + last_part = message.parts[-1] + if ( + message.role == "user" + and last_part.inline_data + and last_part.inline_data.mime_type == "audio/wav" + ): + self._context.messages[-2] = glm.Content( + role="user", parts=[glm.Part(text=self._transcript)] + ) + + def add_transcript_back_to_inference_output(self): + if not self._transcript: + return + message = self._context.messages[-1] + last_part = message.parts[-1] + if message.role == "model" and last_part.text: + self._context.messages[-1].parts[-1].text += f"\n\n{marker}\n{self._transcript}\n" + + async def process_frame(self, frame, direction): + await super().process_frame(frame, direction) + + if isinstance(frame, MagicDemoTranscriptionFrame): + self._transcript = frame.text + elif isinstance(frame, LLMFullResponseEndFrame) or isinstance( + frame, StartInterruptionFrame + ): + self.swap_user_audio() + self.add_transcript_back_to_inference_output() + self._transcript = "" + + await self.push_frame(frame, direction) + + +async def main(): + async with aiohttp.ClientSession() as session: + (room_url, token) = await configure(session) + + transport = DailyTransport( + room_url, + token, + "Respond bot", + DailyParams( + audio_out_enabled=True, + # No transcription at all. just audio input to Gemini! + # transcription_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + vad_audio_passthrough=True, + ), + ) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady + ) + + llm = GoogleLLMService(model="gemini-1.5-flash-latest", api_key=os.getenv("GOOGLE_API_KEY")) + + messages = [ + { + "role": "system", + "content": system_message, + }, + { + "role": "user", + "content": "Start by saying hello.", + }, + ] + + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) + audio_collector = UserAudioCollector(context, context_aggregator.user()) + pull_transcript_out_of_llm_output = TranscriptExtractor(context) + fixup_context_messages = TanscriptionContextFixup(context) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + audio_collector, + context_aggregator.user(), # User responses + llm, # LLM + pull_transcript_out_of_llm_output, + tts, # TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses + fixup_context_messages, + ] + ) + + task = PipelineTask( + pipeline, + PipelineParams( + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + ), + ) + + @transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + await transport.capture_participant_transcription(participant["id"]) + # Kick off the conversation. + await task.queue_frames([context_aggregator.user().get_context_frame()]) + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/src/pipecat/processors/aggregators/openai_llm_context.py b/src/pipecat/processors/aggregators/openai_llm_context.py index d70f0e25b..5e4f44093 100644 --- a/src/pipecat/processors/aggregators/openai_llm_context.py +++ b/src/pipecat/processors/aggregators/openai_llm_context.py @@ -15,6 +15,7 @@ from loguru import logger from PIL import Image from pipecat.frames.frames import ( + AudioRawFrame, Frame, FunctionCallInProgressFrame, FunctionCallResultFrame, @@ -174,6 +175,10 @@ class OpenAILLMContext: content.append({"type": "text", "text": text}) self.add_message({"role": "user", "content": content}) + def add_audio_frames_message(self, *, audio_frames: list[AudioRawFrame], text: str = None): + # todo: implement for OpenAI models and others + pass + async def call_function( self, f: Callable[ @@ -213,6 +218,29 @@ class OpenAILLMContext: await f(function_name, tool_call_id, arguments, llm, self, function_call_result_callback) + def create_wav_header(self, sample_rate, num_channels, bits_per_sample, data_size): + # RIFF chunk descriptor + header = bytearray() + header.extend(b"RIFF") # ChunkID + header.extend((data_size + 36).to_bytes(4, "little")) # ChunkSize: total size - 8 + header.extend(b"WAVE") # Format + # "fmt " sub-chunk + header.extend(b"fmt ") # Subchunk1ID + header.extend((16).to_bytes(4, "little")) # Subchunk1Size (16 for PCM) + header.extend((1).to_bytes(2, "little")) # AudioFormat (1 for PCM) + header.extend(num_channels.to_bytes(2, "little")) # NumChannels + header.extend(sample_rate.to_bytes(4, "little")) # SampleRate + # Calculate byte rate and block align + byte_rate = sample_rate * num_channels * (bits_per_sample // 8) + block_align = num_channels * (bits_per_sample // 8) + header.extend(byte_rate.to_bytes(4, "little")) # ByteRate + header.extend(block_align.to_bytes(2, "little")) # BlockAlign + header.extend(bits_per_sample.to_bytes(2, "little")) # BitsPerSample + # "data" sub-chunk + header.extend(b"data") # Subchunk2ID + header.extend(data_size.to_bytes(4, "little")) # Subchunk2Size + return header + @dataclass class OpenAILLMContextFrame(Frame): diff --git a/src/pipecat/services/google.py b/src/pipecat/services/google.py index c1114e44e..69864262e 100644 --- a/src/pipecat/services/google.py +++ b/src/pipecat/services/google.py @@ -16,6 +16,7 @@ from PIL import Image from pydantic import BaseModel, Field from pipecat.frames.frames import ( + AudioRawFrame, ErrorFrame, Frame, LLMFullResponseEndFrame, @@ -184,11 +185,53 @@ class GoogleLLMContext(OpenAILLMContext): msgs.append(obj) return msgs + def add_image_frame_message( + self, *, format: str, size: tuple[int, int], image: bytes, text: str = None + ): + buffer = io.BytesIO() + Image.frombytes(format, size, image).save(buffer, format="JPEG") + + parts = [] + if text: + parts.append(glm.Part(text=text)) + parts.append( + glm.Part(inline_data=glm.Blob(mime_type="image/jpeg", data=buffer.getvalue())), + ) + self.add_message(glm.Content(role="user", parts=parts)) + + def add_audio_frames_message(self, *, audio_frames: list[AudioRawFrame], text: str = None): + if not audio_frames: + return + + sample_rate = audio_frames[0].sample_rate + num_channels = audio_frames[0].num_channels + + parts = [] + data = b"".join(frame.audio for frame in audio_frames) + if text: + parts.append(glm.Part(text=text)) + parts.append( + glm.Part( + inline_data=glm.Blob( + mime_type="audio/wav", + data=( + bytes( + self.create_wav_header(sample_rate, num_channels, 16, len(data)) + data + ) + ), + ) + ), + ) + self.add_message(glm.Content(role="user", parts=parts)) + # message = {"mime_type": "audio/mp3", "data": bytes(data + create_wav_header(sample_rate, num_channels, 16, len(data)))} + # self.add_message(message) + def from_standard_message(self, message): role = message["role"] content = message.get("content", []) if role == "system": - role = "user" + self.system_message = content + return None elif role == "assistant": role = "model" @@ -232,20 +275,6 @@ class GoogleLLMContext(OpenAILLMContext): message = glm.Content(role=role, parts=parts) return message - def add_image_frame_message( - self, *, format: str, size: tuple[int, int], image: bytes, text: str = None - ): - buffer = io.BytesIO() - Image.frombytes(format, size, image).save(buffer, format="JPEG") - - parts = [] - if text: - parts.append(glm.Part(text=text)) - parts.append( - glm.Part(inline_data=glm.Blob(mime_type="image/jpeg", data=buffer.getvalue())), - ) - self.add_message(glm.Content(role="user", parts=parts)) - def to_standard_messages(self, obj) -> list: msg = {"role": obj.role, "content": []} if msg["role"] == "model": @@ -289,9 +318,20 @@ class GoogleLLMContext(OpenAILLMContext): return [msg] def _restructure_from_openai_messages(self): + self.system_message = None # first, map across self._messages calling self.from_standard_message(m) to modify messages in place try: - self._messages[:] = [self.from_standard_message(m) for m in self._messages] + self._messages[:] = [ + msg + for msg in (self.from_standard_message(m) for m in self._messages) + if msg is not None + ] + # We might have been given a messages list with only a system message. If so, let's put that back in + # the messages list as a user message. + if self.system_message and not self._messages: + self.add_message( + glm.Content(role="user", parts=[glm.Part(text=self.system_message)]) + ) except Exception as e: logger.error(f"Error mapping messages: {e}") # iterate over messages and remove any messages that have an empty content list @@ -319,11 +359,14 @@ class GoogleLLMService(LLMService): api_key: str, model: str = "gemini-1.5-flash-latest", params: InputParams = InputParams(), + system_instruction: Optional[str] = None, **kwargs, ): super().__init__(**kwargs) gai.configure(api_key=api_key) - self._create_client(model) + self.set_model_name(model) + self._system_instruction = system_instruction + self._create_client() self._settings = { "max_tokens": params.max_tokens, "temperature": params.temperature, @@ -335,34 +378,10 @@ class GoogleLLMService(LLMService): def can_generate_metrics(self) -> bool: return True - def _create_client(self, model: str): - self.set_model_name(model) - self._client = gai.GenerativeModel(model) - - def _get_messages_from_openai_context(self, context: OpenAILLMContext) -> List[glm.Content]: - openai_messages = context.get_messages() - google_messages = [] - - for message in openai_messages: - role = message["role"] - content = message["content"] - if role == "system": - role = "user" - elif role == "assistant": - role = "model" - - parts = [glm.Part(text=content)] - if "mime_type" in message: - parts.append( - glm.Part( - inline_data=glm.Blob( - mime_type=message["mime_type"], data=message["data"].getvalue() - ) - ) - ) - google_messages.append({"role": role, "parts": parts}) - - return google_messages + def _create_client(self): + self._client = gai.GenerativeModel( + self._model_name, system_instruction=self._system_instruction + ) async def _async_generator_wrapper(self, sync_generator): for item in sync_generator: @@ -374,10 +393,11 @@ class GoogleLLMService(LLMService): try: logger.debug(f"Generating chat: {context.get_messages_for_logging()}") - # todo: move this into the new context code structure, convert from openai context one time - # todo: add system instructions - # messages = self._get_messages_from_openai_context(context) messages = context.messages + if self._system_instruction != context.system_message: + logger.debug(f"System instruction changed: {context.system_message}") + self._system_instruction = context.system_message + self._create_client() # Filter out None values and create GenerationConfig generation_params = { @@ -394,24 +414,21 @@ class GoogleLLMService(LLMService): generation_config = GenerationConfig(**generation_params) if generation_params else None await self.start_ttfb_metrics() - tools = context.tools if context.tools else [] response = self._client.generate_content( contents=messages, tools=tools, stream=True, generation_config=generation_config ) - - tokens = LLMTokenUsage( - prompt_tokens=response.usage_metadata.prompt_token_count, - completion_tokens=response.usage_metadata.candidates_token_count, - total_tokens=response.usage_metadata.total_token_count, - ) - - await self.start_llm_usage_metrics(tokens) - await self.stop_ttfb_metrics() + prompt_tokens = response.usage_metadata.prompt_token_count + completion_tokens = response.usage_metadata.candidates_token_count + total_tokens = response.usage_metadata.total_token_count + async for chunk in self._async_generator_wrapper(response): - # todo: usage + if chunk.usage_metadata: + prompt_tokens += response.usage_metadata.prompt_token_count + completion_tokens += response.usage_metadata.candidates_token_count + total_tokens += response.usage_metadata.total_token_count try: for c in chunk.parts: if c.text: @@ -436,6 +453,13 @@ class GoogleLLMService(LLMService): except Exception as e: logger.exception(f"{self} exception: {e}") finally: + await self.start_llm_usage_metrics( + LLMTokenUsage( + prompt_tokens=prompt_tokens, + completion_tokens=completion_tokens, + total_tokens=total_tokens, + ) + ) await self.push_frame(LLMFullResponseEndFrame()) async def process_frame(self, frame: Frame, direction: FrameDirection): From 335178ff063b2275fe283c39c80d141996e99c2b Mon Sep 17 00:00:00 2001 From: Kwindla Hultman Kramer Date: Mon, 11 Nov 2024 21:04:15 -0800 Subject: [PATCH 8/8] some gemini audio input examples --- .../07p-interruptible-google-audio-in.py | 6 +- .../22d-natural-conversation-gemini-audio.py | 355 ++++++++++++++++++ 2 files changed, 357 insertions(+), 4 deletions(-) create mode 100644 examples/foundational/22d-natural-conversation-gemini-audio.py diff --git a/examples/foundational/07p-interruptible-google-audio-in.py b/examples/foundational/07p-interruptible-google-audio-in.py index 33bb30187..40389274a 100644 --- a/examples/foundational/07p-interruptible-google-audio-in.py +++ b/examples/foundational/07p-interruptible-google-audio-in.py @@ -55,7 +55,7 @@ Each time you answer, you should respond in three parts. 1. Transcribe exactly what the user said. 2. Output the separator field '{marker}'. -3. Respond to the user's input in a succinct, helpful, creative way using only simple text and punctuation. +3. Respond to the user's input in a helpful, creative way using only simple text and punctuation. Example: @@ -78,9 +78,7 @@ class UserAudioCollector(FrameProcessor): self._context = context self._user_context_aggregator = user_context_aggregator self._audio_frames = [] - self._start_secs = ( - 0.2 # this should match VAD_START_SECS but we'll just hardcode it for now - ) + self._start_secs = 0.2 # this should match VAD start_secs (hardcoding for now) self._user_speaking = False async def process_frame(self, frame, direction): diff --git a/examples/foundational/22d-natural-conversation-gemini-audio.py b/examples/foundational/22d-natural-conversation-gemini-audio.py new file mode 100644 index 000000000..1ff8aa23e --- /dev/null +++ b/examples/foundational/22d-natural-conversation-gemini-audio.py @@ -0,0 +1,355 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import aiohttp +import asyncio +import os +import sys +import time + +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.frames.frames import LLMMessagesFrame, TextFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.parallel_pipeline import ParallelPipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.services.deepgram import DeepgramSTTService +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.openai_llm_context import ( + OpenAILLMContext, +) +from pipecat.services.cartesia import CartesiaTTSService +from pipecat.services.google import GoogleLLMService, GoogleLLMContext +from pipecat.sync.event_notifier import EventNotifier +from pipecat.transports.services.daily import DailyParams, DailyTransport +from pipecat.processors.frame_processor import FrameProcessor, FrameDirection +from pipecat.frames.frames import ( + CancelFrame, + EndFrame, + Frame, + InputAudioRawFrame, + StartFrame, + StartInterruptionFrame, + StopInterruptionFrame, + SystemFrame, + TranscriptionFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, +) +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContextFrame +from pipecat.sync.base_notifier import BaseNotifier +from pipecat.processors.filters.function_filter import FunctionFilter +from pipecat.processors.user_idle_processor import UserIdleProcessor + + +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") + + +classifier_statement = """You are an audio language classifier model. You are receiving audio from a user in a WebRTC call. Your job is to decide whether the user has finished speaking or not. + +Categorize the input you receive as either: + +1. a complete thought, statement, or question, or +2. an incomplete thought, statement, or question + +Output 'YES' if the input is likely to be a completed thought, statement, or question. + +Output 'NO' if the input indicates that the user is still speaking and does not yet expect a response yet. + +If you are unsure, output 'YES'. +""" + +conversational_system_message = """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. + +Please be very concise in your responses. Unless you are explicitly asked to do otherwise, give me the shortest complete answer possible without unnecessary elaboration. Generally you should answer with a single sentence. +""" + + +class StatementJudgeAudioContextAccumulator(FrameProcessor): + def __init__(self, *, notifier: BaseNotifier, **kwargs): + super().__init__(**kwargs) + self._notifier = notifier + self._audio_frames = [] + self._audio_frames = [] + self._start_secs = 0.2 # this should match VAD start_secs (hardcoding for now) + self._user_speaking = False + + async def reset(self): + self._audio_frames = [] + self._user_speaking = False + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + # ignore context frame + if isinstance(frame, OpenAILLMContextFrame): + return + + if isinstance(frame, TranscriptionFrame): + # We could gracefully handle both audio input and text/transcription input ... + # but let's leave that as an exercise to the reader. :-) + return + if isinstance(frame, UserStartedSpeakingFrame): + self._user_speaking = True + elif isinstance(frame, UserStoppedSpeakingFrame): + self._user_speaking = False + context = GoogleLLMContext() + context.set_messages([{"role": "system", "content": classifier_statement}]) + context.add_audio_frames_message(audio_frames=self._audio_frames) + await self.push_frame(OpenAILLMContextFrame(context=context)) + elif isinstance(frame, InputAudioRawFrame): + if self._user_speaking: + self._audio_frames.append(frame) + else: + # Append the audio frame to our buffer. Treat the buffer as a ring buffer, dropping the oldest + # frames as necessary. Assume all audio frames have the same duration. + self._audio_frames.append(frame) + frame_duration = len(frame.audio) / 16 * frame.num_channels / frame.sample_rate + buffer_duration = frame_duration * len(self._audio_frames) + while buffer_duration > self._start_secs: + self._audio_frames.pop(0) + buffer_duration -= frame_duration + + await self.push_frame(frame, direction) + + +class CompletenessCheck(FrameProcessor): + def __init__( + self, notifier: BaseNotifier, audio_accumulator: StatementJudgeAudioContextAccumulator + ): + super().__init__() + self._notifier = notifier + self._audio_accumulator = audio_accumulator + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if isinstance(frame, TextFrame) and frame.text.startswith("YES"): + logger.debug("Completeness check YES") + await self.push_frame(UserStoppedSpeakingFrame()) + await self._audio_accumulator.reset() + await self._notifier.notify() + elif isinstance(frame, TextFrame): + if frame.text.strip(): + logger.debug(f"Completeness check NO - '{frame.text}'") + + +class OutputGate(FrameProcessor): + def __init__(self, notifier: BaseNotifier, **kwargs): + super().__init__(**kwargs) + self._gate_open = False + self._frames_buffer = [] + self._notifier = notifier + + def close_gate(self): + self._gate_open = False + + def open_gate(self): + self._gate_open = True + + 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): + if isinstance(frame, StartFrame): + await self._start() + if isinstance(frame, (EndFrame, CancelFrame)): + await self._stop() + if isinstance(frame, StartInterruptionFrame): + self._frames_buffer = [] + self.close_gate() + await self.push_frame(frame, direction) + return + + # Ignore frames that are not following the direction of this gate. + if direction != FrameDirection.DOWNSTREAM: + await self.push_frame(frame, direction) + return + + if self._gate_open: + await self.push_frame(frame, direction) + return + + self._frames_buffer.append((frame, direction)) + + async def _start(self): + self._frames_buffer = [] + self._gate_task = self.get_event_loop().create_task(self._gate_task_handler()) + + async def _stop(self): + self._gate_task.cancel() + await self._gate_task + + async def _gate_task_handler(self): + while True: + try: + await self._notifier.wait() + self.open_gate() + for frame, direction in self._frames_buffer: + await self.push_frame(frame, direction) + self._frames_buffer = [] + except asyncio.CancelledError: + break + + +async def main(): + async with aiohttp.ClientSession() as session: + (room_url, _) = await configure(session) + + transport = DailyTransport( + room_url, + None, + "Respond bot", + DailyParams( + audio_out_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + vad_audio_passthrough=True, + ), + ) + + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady + ) + + # This is the LLM that will be used to detect if the user has finished a + # statement. This doesn't really need to be an LLM, we could use NLP + # libraries for that, but we have the machinery to use an LLM, so we might as well! + statement_llm = GoogleLLMService( + model="gemini-1.5-flash-latest", api_key=os.getenv("GOOGLE_API_KEY") + ) + + # This is the regular LLM. + llm = GoogleLLMService(model="gemini-1.5-flash-latest", api_key=os.getenv("GOOGLE_API_KEY")) + + messages = [ + { + "role": "system", + "content": conversational_system_message, + }, + ] + + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) + + # We have instructed the LLM to return 'YES' if it thinks the user + # completed a sentence. So, if it's 'YES' we will return true in this + # predicate which will wake up the notifier. + async def wake_check_filter(frame): + return frame.text == "YES" + + # This is a notifier that we use to synchronize the two LLMs. + notifier = EventNotifier() + + # This turns the LLM context into an inference request to classify the user's speech + # as complete or incomplete. + statement_judge_context_filter = StatementJudgeAudioContextAccumulator(notifier=notifier) + + # This sends a UserStoppedSpeakingFrame and triggers the notifier event + completeness_check = CompletenessCheck( + notifier=notifier, audio_accumulator=statement_judge_context_filter + ) + + # # Notify if the user hasn't said anything. + async def user_idle_notifier(frame): + await notifier.notify() + + # Sometimes the LLM will fail detecting if a user has completed a + # sentence, this will wake up the notifier if that happens. + user_idle = UserIdleProcessor(callback=user_idle_notifier, timeout=5.0) + + bot_output_gate = OutputGate(notifier=notifier) + + async def block_user_stopped_speaking(frame): + return not isinstance(frame, UserStoppedSpeakingFrame) + + async def pass_only_llm_trigger_frames(frame): + return ( + isinstance(frame, OpenAILLMContextFrame) + or isinstance(frame, LLMMessagesFrame) + or isinstance(frame, StartInterruptionFrame) + or isinstance(frame, StopInterruptionFrame) + ) + + pipeline = Pipeline( + [ + transport.input(), + ParallelPipeline( + [ + # Pass everything except UserStoppedSpeaking to the elements after + # this ParallelPipeline + FunctionFilter(filter=block_user_stopped_speaking), + ], + [ + statement_judge_context_filter, + statement_llm, + completeness_check, + ], + [ + stt, + context_aggregator.user(), + # Block everything except OpenAILLMContextFrame and LLMMessagesFrame + FunctionFilter(filter=pass_only_llm_trigger_frames), + llm, + bot_output_gate, # Buffer all llm/tts output until notified. + ], + ), + tts, + user_idle, + transport.output(), + context_aggregator.assistant(), + ] + ) + + task = PipelineTask( + pipeline, + PipelineParams( + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + ), + ) + + @transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + await transport.capture_participant_transcription(participant["id"]) + # Kick off the conversation. + await task.queue_frames([context_aggregator.user().get_context_frame()]) + + @transport.event_handler("on_app_message") + async def on_app_message(transport, message, sender): + logger.debug(f"Received app message: {message} - {sender}") + if "message" not in message: + return + + await task.queue_frames( + [ + UserStartedSpeakingFrame(), + TranscriptionFrame( + user_id=sender, timestamp=time.time(), text=message["message"] + ), + UserStoppedSpeakingFrame(), + ] + ) + + runner = PipelineRunner() + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main())