diff --git a/examples/foundational/39-aws-nova-sonic.py b/examples/foundational/39-aws-nova-sonic.py index f08cfad04..07670f75a 100644 --- a/examples/foundational/39-aws-nova-sonic.py +++ b/examples/foundational/39-aws-nova-sonic.py @@ -83,12 +83,16 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): ), ) - # Specify initial system instruction + # Specify initial system instruction. + # HACK: note that, for now, we need to inject a special bit of text into this instruction to + # allow the first assistant response to be programmatically triggered (which happens in the + # on_client_connected handler, below) # TODO: looks like Nova Sonic can't handle new lines? system_instruction = ( - "You are a friendly assistant. The user and you will engage in a spoken dialog " - "exchanging the transcripts of a natural real-time conversation. Keep your responses short, " - "generally two or three sentences for chatty scenarios." + "You are a friendly assistant. The user and you will engage in a spoken dialog exchanging " + "the transcripts of a natural real-time conversation. Keep your responses short, generally " + "two or three sentences for chatty scenarios. " + f"{AWSNovaSonicLLMService.AWAIT_TRIGGER_ASSISTANT_RESPONSE_INSTRUCTION}" ) # Create the AWS Nova Sonic LLM service @@ -117,7 +121,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): {"role": "system", "content": f"{system_instruction}"}, { "role": "user", - "content": "Say hello!", + "content": "Tell me a fun fact!", }, ], tools=tools, @@ -151,6 +155,10 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): logger.info(f"Client connected") # Kick off the conversation. await task.queue_frames([context_aggregator.user().get_context_frame()]) + # HACK: for now, we need this special way of triggering the first assistant response in AWS + # Nova Sonic. Note that this trigger requires a special corresponding bit of text in the + # system instruction. In the future, simply queueing the context frame should be sufficient. + await llm.trigger_assistant_response() # Handle client disconnection events @transport.event_handler("on_client_disconnected") diff --git a/pyproject.toml b/pyproject.toml index d6d05c00c..7ce167d77 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -96,6 +96,7 @@ where = ["src"] [tool.setuptools.package-data] "pipecat" = ["py.typed"] +"pipecat.services.aws_nova_sonic" = ["src/pipecat/services/aws_nova_sonic/ready.wav"] [tool.pytest.ini_options] addopts = "--verbose" diff --git a/src/pipecat/services/aws_nova_sonic/aws.py b/src/pipecat/services/aws_nova_sonic/aws.py index e7b1fd8e6..5b69810f3 100644 --- a/src/pipecat/services/aws_nova_sonic/aws.py +++ b/src/pipecat/services/aws_nova_sonic/aws.py @@ -8,8 +8,10 @@ import asyncio import base64 import json import uuid +import wave from dataclasses import dataclass from enum import Enum +from importlib.resources import files from typing import Any, List, Optional from aws_sdk_bedrock_runtime.client import ( @@ -146,6 +148,8 @@ class AWSNovaSonicLLMService(LLMService): self._context_available = False self._ready_to_send_context = False self._handling_bot_stopped_speaking = False + self._triggering_assistant_response = False + self._assistant_response_trigger_audio: bytes = None # Not cleared on _disconnect() # # standard AIService frame handling @@ -180,8 +184,7 @@ class AWSNovaSonicLLMService(LLMService): if isinstance(frame, OpenAILLMContextFrame): await self._handle_context(frame.context) elif isinstance(frame, InputAudioRawFrame): - # TODO: check if _audio_input_paused? what causes that? - await self._send_user_audio_event(frame) + await self._handle_input_audio_frame(frame) elif isinstance(frame, BotStoppedSpeakingFrame): await self._handle_bot_stopped_speaking() elif isinstance(frame, AWSNovaSonicFunctionCallResultFrame): @@ -211,6 +214,15 @@ class AWSNovaSonicLLMService(LLMService): self._context_available = True await self._finish_connecting_if_context_available() + async def _handle_input_audio_frame(self, frame: InputAudioRawFrame): + # Wait until we're done sending the assistant response trigger audio before sending audio + # from the user's mic + if self._triggering_assistant_response: + return + + # TODO: check if _audio_input_paused? what causes that? + await self._send_user_audio_event(frame.audio) + async def _handle_bot_stopped_speaking(self): # Protect against back-to-back BotStoppedSpeaking calls, which I've observed if self._handling_bot_stopped_speaking: @@ -316,6 +328,14 @@ class AWSNovaSonicLLMService(LLMService): # Start receiving events self._receive_task = self.create_task(self._receive_task_handler()) + # If we need to, send assistant response trigger + if self._triggering_assistant_response: + # If the trigger was the first audio chunk sent on this connection it'd be ignored (I'm + # guessing the LLM can't quite "hear" the first little bit of audio sent). So send a bit + # of leading blank audio first. + await self._send_assistant_response_trigger(lead_with_blank_audio=True) + self._triggering_assistant_response = False + async def _disconnect(self): try: # Clean up receive task @@ -340,6 +360,8 @@ class AWSNovaSonicLLMService(LLMService): self._assistant_is_responding = False self._context_available = False self._ready_to_send_context = False + self._handling_bot_stopped_speaking = False + self._triggering_assistant_response = False except Exception as e: logger.error(f"{self} error disconnecting: {e}") @@ -490,11 +512,11 @@ class AWSNovaSonicLLMService(LLMService): ''' await self._send_client_event(text_content_end) - async def _send_user_audio_event(self, frame: InputAudioRawFrame): + async def _send_user_audio_event(self, audio: bytes): if not self._stream: return - blob = base64.b64encode(frame.audio) + blob = base64.b64encode(audio) audio_event = f''' {{ "event": {{ @@ -639,7 +661,7 @@ class AWSNovaSonicLLMService(LLMService): elif "contentEnd" in event_json: # Handle a piece of content ending await self._handle_content_end_event(event_json) - elif "completionStart" in event_json: + elif "completionEnd" in event_json: # Handle the LLM completion ending await self._handle_completion_end_event(event_json) @@ -839,7 +861,7 @@ class AWSNovaSonicLLMService(LLMService): ) # - # Context + # context # def create_context_aggregator( @@ -855,3 +877,61 @@ class AWSNovaSonicLLMService(LLMService): assistant = AWSNovaSonicAssistantContextAggregator(context=context, params=assistant_params) return AWSNovaSonicContextAggregatorPair(user, assistant) + + # + # assistant response trigger (HACK) + # + + # Class variable + AWAIT_TRIGGER_ASSISTANT_RESPONSE_INSTRUCTION = ( + "Start speaking when you hear the user say 'ready', but don't consider that 'ready' to be " + "a meaningful part of the conversation other than as a trigger for you to start speaking." + ) + + async def trigger_assistant_response(self): + if self._triggering_assistant_response: + return False + + self._triggering_assistant_response = True + + # Read audio bytes, if we don't already have them cached + if not self._assistant_response_trigger_audio: + file_path = files("pipecat.services.aws_nova_sonic").joinpath("ready.wav") + with wave.open(file_path.open("rb"), "rb") as wav_file: + self._assistant_response_trigger_audio = wav_file.readframes(wav_file.getnframes()) + + # Send the trigger audio, if we're fully connected and set up + # NOTE: maybe there's a better way to determine whether we're done setting up? + if self._receive_task: + await self._send_assistant_response_trigger() + self._triggering_assistant_response = False + + async def _send_assistant_response_trigger(self, lead_with_blank_audio=False): + # TODO: if/when we make bitrate, etc configurable, avoid hard-coding this + chunk_size = 640 # equivalent to what we get from InputAudioRawFrame + chunk_duration = 640 / ( + 16000 * 2 + ) # 640 bytes of 16-bit (2-byte) PCM mono audio at 16kHz corresponds to 0.02 seconds + + # Lead with blank audio, if needed + if lead_with_blank_audio: + blank_audio_duration = 0.5 # much less than this and it doesn't reliably work + blank_audio_chunk = b"\x00" * chunk_size + num_chunks = int(blank_audio_duration / chunk_duration) + for _ in range(num_chunks): + await self._send_user_audio_event(blank_audio_chunk) + await asyncio.sleep(chunk_duration) + + # Send trigger audio + # NOTE: this audio *will* be transcribed and eventually make it into the context. That's OK: + # if we ever need to seed this service again with context it would make sense to include it + # since the instruction (i.e. the "wait for the trigger" instruction) will be part of the + # context as well. + # print(f"[pk] sending trigger audio! {len(self._assistant_response_trigger_audio)}") + audio_chunks = [ + self._assistant_response_trigger_audio[i : i + chunk_size] + for i in range(0, len(self._assistant_response_trigger_audio), chunk_size) + ] + for chunk in audio_chunks: + await self._send_user_audio_event(chunk) + await asyncio.sleep(chunk_duration) diff --git a/src/pipecat/services/aws_nova_sonic/ready.wav b/src/pipecat/services/aws_nova_sonic/ready.wav new file mode 100644 index 000000000..ca932afa6 Binary files /dev/null and b/src/pipecat/services/aws_nova_sonic/ready.wav differ