diff --git a/examples/foundational/39-aws-nova-sonic.py b/examples/foundational/39-aws-nova-sonic.py index fd7568d63..fffaee686 100644 --- a/examples/foundational/39-aws-nova-sonic.py +++ b/examples/foundational/39-aws-nova-sonic.py @@ -9,6 +9,7 @@ import os from dotenv import load_dotenv from loguru import logger +# import logging from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.audio.vad.vad_analyzer import VADParams from pipecat.frames.frames import LLMMessagesAppendFrame @@ -23,6 +24,11 @@ from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection # Load environment variables load_dotenv(override=True) +# logging.basicConfig( +# level=logging.DEBUG, +# format='%(asctime)s - %(levelname)s - %(message)s' +# ) + async def run_bot(webrtc_connection: SmallWebRTCConnection): logger.info(f"Starting bot") @@ -51,9 +57,11 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): # Respond to what the user said in a creative and helpful way. # """ # 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." + 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." + ) llm = AWSNovaSonicService( instruction=system_instruction, diff --git a/src/pipecat/services/aws_nova_sonic/aws.py b/src/pipecat/services/aws_nova_sonic/aws.py index aff0be2d2..f8517023e 100644 --- a/src/pipecat/services/aws_nova_sonic/aws.py +++ b/src/pipecat/services/aws_nova_sonic/aws.py @@ -94,7 +94,7 @@ class AWSNovaSonicService(LLMService): await self.push_frame(frame, direction) # - # communication with LLM + # LLM communication: lifecycle # async def _connect(self): @@ -144,6 +144,10 @@ class AWSNovaSonicService(LLMService): ) return BedrockRuntimeClient(config=config) + # + # LLM communication: input events (pipecat -> LLM) + # + # TODO: make params configurable? async def _send_session_start(self): session_start = """ @@ -278,6 +282,18 @@ class AWSNovaSonicService(LLMService): ) await self._stream.input_stream.send(event) + # + # LLM communication: output events (LLM -> pipecat) + # + + # Receive LLM responses ("completions"). + # Each response contains up to four pieces of content, delivered sequentially: + # - User transcription + # - Tool use (optional) + # - Text response + # - Audio response + # Each piece of content is wrapped by "contentStart" and "contentEnd" events. + # Each overall response is wrapped by "completionStart" and "completionEnd" events. async def _receive_task_handler(self): try: while self._client: @@ -288,13 +304,46 @@ class AWSNovaSonicService(LLMService): response_data = result.value.bytes_.decode("utf-8") json_data = json.loads(response_data) - if "audioOutput" in json_data["event"]: - await self._handle_audio_output_event(json_data["event"]) + if "event" in json_data: + event_json = json_data["event"] + if "completionStart" in event_json: + # Handle the LLM response starting + await self._handle_completion_start_event(event_json) + elif "contentStart" in event_json: + # Handle a piece of content starting + await self._handle_content_start_event(event_json) + elif "textOutput" in event_json: + # Handle text output content + await self._handle_text_output_event(event_json) + elif "audioOutput" in event_json: + # Handle audio output content + await self._handle_audio_output_event(event_json) + elif "contentEnd" in event_json: + # Handle a piece of content ending + await self._handle_content_end_event(event_json) + elif "completionStart" in event_json: + # Handle the LLM response ending + await self._handle_completion_end_event(event_json) + except Exception as e: logger.error(f"{self} error processing responses: {e}") + async def _handle_completion_start_event(self, event_json): + print("[pk] completion start") + + async def _handle_content_start_event(self, event_json): + content_start = event_json["contentStart"] + type = content_start["type"] + role = content_start["role"] + print(f"[pk] content start. type: {type}, role: {role}") + + async def _handle_text_output_event(self, event_json): + text_content = event_json["textOutput"]["content"] + print(f"[pk] text output. content: {text_content}") + async def _handle_audio_output_event(self, event_json): audio_content = event_json["audioOutput"]["content"] + print(f"[pk] audio output. content: {len(audio_content)}") audio = base64.b64decode(audio_content) # TODO: how is _current_audio_response used? # TODO: make sample rate + channels (used in multiple places) consts @@ -304,3 +353,11 @@ class AWSNovaSonicService(LLMService): num_channels=1, ) await self.push_frame(frame) + + async def _handle_content_end_event(self, event_json): + content_end = event_json["contentEnd"] + type = content_end["type"] + print(f"[pk] content end. type: {type}") + + async def _handle_completion_end_event(self, event_json): + print("[pk] completion end")