diff --git a/CHANGELOG.md b/CHANGELOG.md index 5c8603121..ec717db58 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,9 +22,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Improving the latency of the `HeyGenVideoService`. -- Updated `15-switch-voices.py` and `15a-switch-languages.py` examples to show - how to enclose complex logic (e.g. `ParallelPipeline`) into a single processor - so the main pipeline becomes simpler. +- Updated foundational examples to show how to enclose complex logic + (e.g. `ParallelPipeline`) into a single processor so the main pipeline becomes + simpler. ## [0.0.79] - 2025-08-07 diff --git a/examples/foundational/22-natural-conversation.py b/examples/foundational/22-natural-conversation.py index 42dbbe8c5..675e06de6 100644 --- a/examples/foundational/22-natural-conversation.py +++ b/examples/foundational/22-natural-conversation.py @@ -25,7 +25,8 @@ from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport from pipecat.services.cartesia.tts import CartesiaTTSService from pipecat.services.deepgram.stt import DeepgramSTTService -from pipecat.services.openai.llm import OpenAILLMService +from pipecat.services.llm_service import LLMService +from pipecat.services.openai.llm import OpenAIContextAggregatorPair, OpenAILLMService from pipecat.sync.event_notifier import EventNotifier from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams @@ -34,6 +35,76 @@ from pipecat.transports.services.daily import DailyParams load_dotenv(override=True) +class TurnDetectionLLM(Pipeline): + def __init__(self, llm: LLMService, context_aggregator: OpenAIContextAggregatorPair): + # 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")) + + 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) + + # 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. + completness_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. We start with the gate open because we send an + # initial context frame to start the conversation. + gated_context_aggregator = GatedOpenAILLMContextAggregator( + notifier=notifier, start_open=True + ) + + # 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) + + # 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. + super().__init__( + [ + ParallelPipeline( + [ + statement_context_aggregator.user(), + statement_llm, + completness_check, + NullFilter(), + ], + [context_aggregator.user(), gated_context_aggregator, llm], + ), + user_idle, + ] + ) + + # We store functions so objects (e.g. SileroVADAnalyzer) don't get # instantiated. The function will be called when the desired transport gets # selected. @@ -66,24 +137,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading 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")) - - 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")) + llm_main = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { @@ -93,53 +148,16 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): ] context = OpenAILLMContext(messages) - context_aggregator = llm.create_context_aggregator(context) + context_aggregator = llm_main.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" + # LLM + turn detection (with an extra LLM as a judge) + llm = TurnDetectionLLM(llm_main, context_aggregator) - # 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. - completness_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. We start with the gate open because we send an - # initial context frame to start the conversation. - gated_context_aggregator = GatedOpenAILLMContextAggregator(notifier=notifier, start_open=True) - - # 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) - - # 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(), # Transport user input - stt, - ParallelPipeline( - [ - statement_context_aggregator.user(), - statement_llm, - completness_check, - NullFilter(), - ], - [context_aggregator.user(), gated_context_aggregator, llm], - ), - user_idle, + stt, # STT + llm, # LLM with turn detection tts, # TTS transport.output(), # Transport bot output context_aggregator.assistant(), # Assistant spoken responses diff --git a/examples/foundational/22b-natural-conversation-proposal.py b/examples/foundational/22b-natural-conversation-proposal.py index 90594cf09..599f91043 100644 --- a/examples/foundational/22b-natural-conversation-proposal.py +++ b/examples/foundational/22b-natural-conversation-proposal.py @@ -6,7 +6,6 @@ import asyncio import os -import time from dotenv import load_dotenv from loguru import logger @@ -44,13 +43,14 @@ from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport from pipecat.services.cartesia.tts import CartesiaTTSService from pipecat.services.deepgram.stt import DeepgramSTTService -from pipecat.services.llm_service import FunctionCallParams +from pipecat.services.llm_service import FunctionCallParams, LLMService from pipecat.services.openai.llm import OpenAILLMService from pipecat.sync.base_notifier import BaseNotifier from pipecat.sync.event_notifier import EventNotifier from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams from pipecat.transports.services.daily import DailyParams +from pipecat.utils.time import time_now_iso8601 load_dotenv(override=True) @@ -192,6 +192,75 @@ async def fetch_weather_from_api(params: FunctionCallParams): await params.result_callback({"conditions": "nice", "temperature": "75"}) +class TurnDetectionLLM(Pipeline): + def __init__(self, llm: LLMService): + # 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 = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) + + # 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() + + # 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) + + # We start with the gate open because we send an initial context frame + # to start the conversation. + bot_output_gate = OutputGate(notifier=notifier, start_open=True) + + async def pass_only_llm_trigger_frames(frame): + return ( + isinstance(frame, OpenAILLMContextFrame) + or isinstance(frame, StartInterruptionFrame) + or isinstance(frame, StopInterruptionFrame) + or isinstance(frame, FunctionCallInProgressFrame) + or isinstance(frame, FunctionCallResultFrame) + ) + + super().__init__( + [ + ParallelPipeline( + [ + # Ignore everything except an OpenAILLMContextFrame. Pass a specially constructed + # simplified context frame 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 frames that trigger LLM inference. + FunctionFilter(filter=pass_only_llm_trigger_frames), + llm, + bot_output_gate, # Buffer all llm/tts output until notified. + ], + ), + user_idle, + ] + ) + + # We store functions so objects (e.g. SileroVADAnalyzer) don't get # instantiated. The function will be called when the desired transport gets # selected. @@ -224,18 +293,13 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading 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 = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) - # This is the regular LLM. - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) + llm_main = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) # You can also register a function_name of None to get all functions # sent to the same callback with an additional function_name parameter. - llm.register_function("get_current_weather", fetch_weather_from_api) + llm_main.register_function("get_current_weather", fetch_weather_from_api) - @llm.event_handler("on_function_calls_started") + @llm_main.event_handler("on_function_calls_started") async def on_function_calls_started(service, function_calls): await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) @@ -272,69 +336,18 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): ] context = OpenAILLMContext(messages, tools) - context_aggregator = llm.create_context_aggregator(context) + context_aggregator = llm_main.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() - - # 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) - - # We start with the gate open because we send an initial context frame - # to start the conversation. - bot_output_gate = OutputGate(notifier=notifier, start_open=True) - - async def pass_only_llm_trigger_frames(frame): - return ( - isinstance(frame, OpenAILLMContextFrame) - or isinstance(frame, StartInterruptionFrame) - or isinstance(frame, StopInterruptionFrame) - or isinstance(frame, FunctionCallInProgressFrame) - or isinstance(frame, FunctionCallResultFrame) - ) + # LLM + turn detection (with an extra LLM as a judge) + llm = TurnDetectionLLM(llm_main) pipeline = Pipeline( [ transport.input(), stt, context_aggregator.user(), - ParallelPipeline( - [ - # Ignore everything except an OpenAILLMContextFrame. Pass a specially constructed - # simplified context frame 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 frames that trigger LLM inference. - FunctionFilter(filter=pass_only_llm_trigger_frames), - llm, - bot_output_gate, # Buffer all llm/tts output until notified. - ], - ), + llm, tts, - user_idle, transport.output(), context_aggregator.assistant(), ] @@ -365,7 +378,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): await task.queue_frames( [ UserStartedSpeakingFrame(), - TranscriptionFrame(user_id="", timestamp=time.time(), text=message["message"]), + TranscriptionFrame( + user_id="", timestamp=time_now_iso8601(), text=message["message"] + ), UserStoppedSpeakingFrame(), ] ) diff --git a/examples/foundational/22c-natural-conversation-mixed-llms.py b/examples/foundational/22c-natural-conversation-mixed-llms.py index c44034f1c..f7a6133b4 100644 --- a/examples/foundational/22c-natural-conversation-mixed-llms.py +++ b/examples/foundational/22c-natural-conversation-mixed-llms.py @@ -6,7 +6,6 @@ import asyncio import os -import time from dotenv import load_dotenv from loguru import logger @@ -45,13 +44,14 @@ from pipecat.runner.utils import create_transport from pipecat.services.anthropic.llm import AnthropicLLMService from pipecat.services.cartesia.tts import CartesiaTTSService from pipecat.services.deepgram.stt import DeepgramSTTService -from pipecat.services.llm_service import FunctionCallParams +from pipecat.services.llm_service import FunctionCallParams, LLMService from pipecat.services.openai.llm import OpenAILLMService from pipecat.sync.base_notifier import BaseNotifier from pipecat.sync.event_notifier import EventNotifier from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams from pipecat.transports.services.daily import DailyParams +from pipecat.utils.time import time_now_iso8601 load_dotenv(override=True) @@ -391,6 +391,75 @@ class OutputGate(FrameProcessor): break +class TurnDetectionLLM(Pipeline): + def __init__(self, llm: LLMService): + # 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")) + + # 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() + + # 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) + + # We start with the gate open because we send an initial context frame + # to start the conversation. + bot_output_gate = OutputGate(notifier=notifier, start_open=True) + + 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, StartInterruptionFrame) + or isinstance(frame, StopInterruptionFrame) + or isinstance(frame, FunctionCallInProgressFrame) + or isinstance(frame, FunctionCallResultFrame) + ) + + super().__init__( + [ + 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 + # simplified context frame 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 frames that trigger LLM inference. + FunctionFilter(filter=pass_only_llm_trigger_frames), + llm, + bot_output_gate, # Buffer all llm/tts output until notified. + ], + ), + user_idle, + ] + ) + + async def fetch_weather_from_api(params: FunctionCallParams): await params.result_callback({"conditions": "nice", "temperature": "75"}) @@ -427,18 +496,13 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading 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")) - # This is the regular LLM. - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) + llm_main = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) # Register a function_name of None to get all functions # sent to the same callback with an additional function_name parameter. - llm.register_function("get_current_weather", fetch_weather_from_api) + llm_main.register_function("get_current_weather", fetch_weather_from_api) - @llm.event_handler("on_function_calls_started") + @llm_main.event_handler("on_function_calls_started") async def on_function_calls_started(service, function_calls): await tts.queue_frame(TTSSpeakFrame("Let me check on that.")) @@ -475,76 +539,18 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): ] context = OpenAILLMContext(messages, tools) - context_aggregator = llm.create_context_aggregator(context) + context_aggregator = llm_main.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 = StatementJudgeContextFilter() - - # 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) - - # We start with the gate open because we send an initial context frame - # to start the conversation. - bot_output_gate = OutputGate(notifier=notifier, start_open=True) - - 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, StartInterruptionFrame) - or isinstance(frame, StopInterruptionFrame) - or isinstance(frame, FunctionCallInProgressFrame) - or isinstance(frame, FunctionCallResultFrame) - ) + # LLM + turn detection (with an extra LLM as a judge) + llm = TurnDetectionLLM(llm_main) 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 - # simplified context frame 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 frames that trigger LLM inference. - FunctionFilter(filter=pass_only_llm_trigger_frames), - llm, - bot_output_gate, # Buffer all llm/tts output until notified. - ], - ), + llm, tts, - user_idle, transport.output(), context_aggregator.assistant(), ] @@ -580,7 +586,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): await task.queue_frames( [ UserStartedSpeakingFrame(), - TranscriptionFrame(user_id="", timestamp=time.time(), text=message["message"]), + TranscriptionFrame( + user_id="", timestamp=time_now_iso8601(), text=message["message"] + ), UserStoppedSpeakingFrame(), ] ) diff --git a/examples/foundational/22d-natural-conversation-gemini-audio.py b/examples/foundational/22d-natural-conversation-gemini-audio.py index 721cfa008..a0871302f 100644 --- a/examples/foundational/22d-natural-conversation-gemini-audio.py +++ b/examples/foundational/22d-natural-conversation-gemini-audio.py @@ -47,11 +47,13 @@ from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport from pipecat.services.cartesia.tts import CartesiaTTSService from pipecat.services.google.llm import GoogleLLMContext, GoogleLLMService +from pipecat.services.llm_service import LLMService from pipecat.sync.base_notifier import BaseNotifier from pipecat.sync.event_notifier import EventNotifier from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams from pipecat.transports.services.daily import DailyParams +from pipecat.utils.time import time_now_iso8601 load_dotenv(override=True) @@ -607,23 +609,90 @@ class OutputGate(FrameProcessor): self._gate_task = None async def _gate_task_handler(self): - while True: - try: - await self._notifier.wait() + await self._notifier.wait() - transcription = await self._transcription_buffer.wait_for_transcription() or "-" - self._context.add_message(Content(role="user", parts=[Part(text=transcription)])) + transcription = await self._transcription_buffer.wait_for_transcription() or "-" + self._context.add_message(Content(role="user", parts=[Part(text=transcription)])) - self.open_gate() - for frame, direction in self._frames_buffer: - await self.push_frame(frame, direction) - self._frames_buffer = [] - except asyncio.CancelledError: - break - except Exception as e: - logger.error(f"OutputGate error: {e}") - raise e - break + self.open_gate() + for frame, direction in self._frames_buffer: + await self.push_frame(frame, direction) + self._frames_buffer = [] + + +class TurnDetectionLLM(Pipeline): + def __init__(self, llm: LLMService, context: OpenAILLMContext): + # This is the LLM that will transcribe user speech. + tx_llm = GoogleLLMService( + name="Transcriber", + model=TRANSCRIBER_MODEL, + api_key=os.getenv("GOOGLE_API_KEY"), + temperature=0.0, + system_instruction=transcriber_system_instruction, + ) + + # This is the LLM that will classify user speech as complete or incomplete. + classifier_llm = GoogleLLMService( + name="Classifier", + model=CLASSIFIER_MODEL, + api_key=os.getenv("GOOGLE_API_KEY"), + temperature=0.0, + system_instruction=classifier_system_instruction, + ) + + # 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) + + audio_accumulater = AudioAccumulator() + # This sends a UserStoppedSpeakingFrame and triggers the notifier event + completeness_check = CompletenessCheck( + notifier=notifier, audio_accumulator=audio_accumulater + ) + + async def block_user_stopped_speaking(frame): + return not isinstance(frame, UserStoppedSpeakingFrame) + + conversation_audio_context_assembler = ConversationAudioContextAssembler(context=context) + + llm_aggregator_buffer = LLMAggregatorBuffer() + + bot_output_gate = OutputGate( + notifier=notifier, context=context, llm_transcription_buffer=llm_aggregator_buffer + ) + + super().__init__( + [ + audio_accumulater, + ParallelPipeline( + [ + # Pass everything except UserStoppedSpeaking to the elements after + # this ParallelPipeline + FunctionFilter(filter=block_user_stopped_speaking), + ], + [ + ParallelPipeline( + [ + classifier_llm, + completeness_check, + ], + [ + tx_llm, + llm_aggregator_buffer, + ], + ) + ], + [ + conversation_audio_context_assembler, + llm, + bot_output_gate, # buffer output until notified, then flush frames and update context + ], + ), + ] + ) # We store functions so objects (e.g. SileroVADAnalyzer) don't get @@ -656,24 +725,6 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady ) - # This is the LLM that will transcribe user speech. - tx_llm = GoogleLLMService( - name="Transcriber", - model=TRANSCRIBER_MODEL, - api_key=os.getenv("GOOGLE_API_KEY"), - temperature=0.0, - system_instruction=transcriber_system_instruction, - ) - - # This is the LLM that will classify user speech as complete or incomplete. - classifier_llm = GoogleLLMService( - name="Classifier", - model=CLASSIFIER_MODEL, - api_key=os.getenv("GOOGLE_API_KEY"), - temperature=0.0, - system_instruction=classifier_system_instruction, - ) - # This is the regular LLM that responds conversationally. conversation_llm = GoogleLLMService( name="Conversation", @@ -685,57 +736,12 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): context = OpenAILLMContext() context_aggregator = conversation_llm.create_context_aggregator(context) - # 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) - - audio_accumulater = AudioAccumulator() - # This sends a UserStoppedSpeakingFrame and triggers the notifier event - completeness_check = CompletenessCheck(notifier=notifier, audio_accumulator=audio_accumulater) - - async def block_user_stopped_speaking(frame): - return not isinstance(frame, UserStoppedSpeakingFrame) - - conversation_audio_context_assembler = ConversationAudioContextAssembler(context=context) - - llm_aggregator_buffer = LLMAggregatorBuffer() - - bot_output_gate = OutputGate( - notifier=notifier, context=context, llm_transcription_buffer=llm_aggregator_buffer - ) + llm = TurnDetectionLLM(conversation_llm, context) pipeline = Pipeline( [ transport.input(), - audio_accumulater, - ParallelPipeline( - [ - # Pass everything except UserStoppedSpeaking to the elements after - # this ParallelPipeline - FunctionFilter(filter=block_user_stopped_speaking), - ], - [ - ParallelPipeline( - [ - classifier_llm, - completeness_check, - ], - [ - tx_llm, - llm_aggregator_buffer, - ], - ) - ], - [ - conversation_audio_context_assembler, - conversation_llm, - bot_output_gate, # buffer output until notified, then flush frames and update context - # TempPrinter(), - ], - ), + llm, tts, transport.output(), context_aggregator.assistant(), @@ -766,7 +772,9 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): await task.queue_frames( [ UserStartedSpeakingFrame(), - TranscriptionFrame(user_id="", timestamp=time.time(), text=message["message"]), + TranscriptionFrame( + user_id="", timestamp=time_now_iso8601(), text=message["message"] + ), UserStoppedSpeakingFrame(), ] )