From 5e5626f04fe59d57fdc4982bcdc7fa48467f7c4f Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 23 Apr 2025 11:40:36 -0400 Subject: [PATCH] [WIP] AWS Nova Sonic service --- examples/foundational/39-aws-nova-sonic.py | 115 ++++++++++++++++++ pyproject.toml | 2 +- .../services/aws_nova_sonic/__init__.py | 1 + src/pipecat/services/aws_nova_sonic/aws.py | 101 +++++++++++++++ 4 files changed, 218 insertions(+), 1 deletion(-) create mode 100644 examples/foundational/39-aws-nova-sonic.py create mode 100644 src/pipecat/services/aws_nova_sonic/__init__.py create mode 100644 src/pipecat/services/aws_nova_sonic/aws.py diff --git a/examples/foundational/39-aws-nova-sonic.py b/examples/foundational/39-aws-nova-sonic.py new file mode 100644 index 000000000..33fbbe477 --- /dev/null +++ b/examples/foundational/39-aws-nova-sonic.py @@ -0,0 +1,115 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import os + +from dotenv import load_dotenv +from loguru import logger + +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.audio.vad.vad_analyzer import VADParams +from pipecat.frames.frames import LLMMessagesAppendFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.services.aws_nova_sonic import AWSNovaSonicService +from pipecat.transports.base_transport import TransportParams +from pipecat.transports.network.small_webrtc import SmallWebRTCTransport +from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection + +# Load environment variables +load_dotenv(override=True) + + +async def run_bot(webrtc_connection: SmallWebRTCConnection): + logger.info(f"Starting bot") + + # Initialize the SmallWebRTCTransport with the connection + transport = SmallWebRTCTransport( + webrtc_connection=webrtc_connection, + params=TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + camera_in_enabled=False, + vad_enabled=True, + vad_audio_passthrough=True, + # set stop_secs to something roughly similar to the internal setting + # of the Multimodal Live api, just to align events. + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.5)), + ), + ) + + # Create the AWS Nova Sonic LLM service + # TODO: system instruction + # system_instruction = f""" + # You are a helpful AI assistant. + # Your goal is to demonstrate your capabilities in a helpful and engaging 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. + # """ + + llm = AWSNovaSonicService( + secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"), + access_key_id=os.getenv("AWS_ACCESS_KEY_ID"), + region=os.getenv("AWS_REGION"), + ) + + # Build the pipeline + pipeline = Pipeline( + [ + transport.input(), + llm, + transport.output(), + ] + ) + + # Configure the pipeline task + task = PipelineTask( + pipeline, + params=PipelineParams( + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + ), + ) + + # Handle client connection event + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info(f"Client connected") + # Kick off the conversation. + await task.queue_frames( + [ + LLMMessagesAppendFrame( + messages=[ + { + "role": "user", + "content": f"Greet the user and introduce yourself.", + } + ] + ) + ] + ) + + # Handle client disconnection events + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info(f"Client disconnected") + + @transport.event_handler("on_client_closed") + async def on_client_closed(transport, client): + logger.info(f"Client closed connection") + await task.cancel() + + # Run the pipeline + runner = PipelineRunner(handle_sigint=False) + await runner.run(task) + + +if __name__ == "__main__": + from run import main + + main() diff --git a/pyproject.toml b/pyproject.toml index 13305933b..d6d05c00c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -41,7 +41,7 @@ Website = "https://pipecat.ai" [project.optional-dependencies] anthropic = [ "anthropic~=0.49.0" ] assemblyai = [ "assemblyai~=0.37.0" ] -aws = [ "boto3~=1.37.16", "websockets~=13.1" ] +aws = [ "boto3~=1.37.16", "websockets~=13.1", "aws_sdk_bedrock_runtime~=0.0.2" ] azure = [ "azure-cognitiveservices-speech~=1.42.0"] cartesia = [ "cartesia~=1.4.0", "websockets~=13.1" ] cerebras = [] diff --git a/src/pipecat/services/aws_nova_sonic/__init__.py b/src/pipecat/services/aws_nova_sonic/__init__.py new file mode 100644 index 000000000..b5559715a --- /dev/null +++ b/src/pipecat/services/aws_nova_sonic/__init__.py @@ -0,0 +1 @@ +from .aws import AWSNovaSonicService diff --git a/src/pipecat/services/aws_nova_sonic/aws.py b/src/pipecat/services/aws_nova_sonic/aws.py new file mode 100644 index 000000000..3caf16761 --- /dev/null +++ b/src/pipecat/services/aws_nova_sonic/aws.py @@ -0,0 +1,101 @@ +from aws_sdk_bedrock_runtime.client import ( + BedrockRuntimeClient, + InvokeModelWithBidirectionalStreamOperationInput, +) +from aws_sdk_bedrock_runtime.config import Config, HTTPAuthSchemeResolver, SigV4AuthScheme +from aws_sdk_bedrock_runtime.models import ( + BidirectionalInputPayloadPart, + InvokeModelWithBidirectionalStreamInput, + InvokeModelWithBidirectionalStreamInputChunk, + InvokeModelWithBidirectionalStreamOperationOutput, + InvokeModelWithBidirectionalStreamOutput, +) +from smithy_aws_core.credentials_resolvers.static import StaticCredentialsResolver +from smithy_aws_core.identity import AWSCredentialsIdentity +from smithy_core.aio.eventstream import DuplexEventStream + +from pipecat.frames.frames import CancelFrame, EndFrame, StartFrame +from pipecat.services.llm_service import LLMService + + +class AWSNovaSonicService(LLMService): + def __init__( + self, + *, + secret_access_key: str, + access_key_id: str, + region: str, + model: str = "amazon.nova-sonic-v1:0", + **kwargs, + ): + super().__init__(**kwargs) + self._secret_access_key = secret_access_key + self._access_key_id = access_key_id + self._region = region + self._model = model + self._client: BedrockRuntimeClient = None + self._stream: DuplexEventStream[ + InvokeModelWithBidirectionalStreamInput, + InvokeModelWithBidirectionalStreamOutput, + InvokeModelWithBidirectionalStreamOperationOutput, + ] = None + self._receive_task = None + + # + # standard AIService frame handling + # + + async def start(self, frame: StartFrame): + await super().start(frame) + await self._connect() + + async def stop(self, frame: EndFrame): + await super().stop(frame) + await self._disconnect() + + async def cancel(self, frame: CancelFrame): + await super().cancel(frame) + await self._disconnect() + + # + # communication + # + + async def _connect(self): + if self._client: + # Here we assume that if we have a client we are connected. + return + self._initialize_client() + self._stream = await self._client.invoke_model_with_bidirectional_stream( + InvokeModelWithBidirectionalStreamOperationInput(model_id=self._model) + ) + self._receive_task = self.create_task(self._receive_task_handler()) + pass + + async def _disconnect(self): + pass + + def _initialize_client(self) -> BedrockRuntimeClient: + config = Config( + endpoint_uri=f"https://bedrock-runtime.{self._region}.amazonaws.com", + region=self._region, + aws_credentials_identity_resolver=StaticCredentialsResolver( + credentials=AWSCredentialsIdentity( + access_key_id=self._access_key_id, + secret_access_key=self._secret_access_key, + # TODO: add additional stuff like aws_session_token + ) + ), + http_auth_scheme_resolver=HTTPAuthSchemeResolver(), + http_auth_schemes={"aws.auth#sigv4": SigV4AuthScheme()}, + ) + self._client = BedrockRuntimeClient(config=config) + + async def _send_client_event(self, event_json): + event = InvokeModelWithBidirectionalStreamInputChunk( + value=BidirectionalInputPayloadPart(bytes_=event_json.encode("utf-8")) + ) + await self._stream.input_stream.send(event) + + async def _receive_task_handler(self): + pass