Merge pull request #1704 from pipecat-ai/pk/amazon-nova-sonic

Amazon Nova Sonic LLM service

CHANGELOG.md

@@ -9,6 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added support for the AWS Nova Sonic speech-to-speech model with the new
+  `AWSNovaSonicLLMService`.
+  See https://docs.aws.amazon.com/nova/latest/userguide/speech.html.
+  Note that it requires Python >= 3.12 and `pip install pipecat-ai[aws-nova-sonic]`.
+
 - Added new AWS services `AWSBedrockLLMService` and `AWSTranscribeSTTService`.
 
 - Added `on_active_speaker_changed` event handler to the `DailyTransport` class.

examples/foundational/20e-persistent-context-aws-nova-sonic.py

@@ -0,0 +1,267 @@
+#
+# Copyright (c) 2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import argparse
+import asyncio
+import glob
+import json
+import os
+from datetime import datetime
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.adapters.schemas.function_schema import FunctionSchema
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.audio.vad.vad_analyzer import VADParams
+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.aws_nova_sonic.aws import AWSNovaSonicLLMService
+from pipecat.services.llm_service import FunctionCallParams
+from pipecat.transports.base_transport import TransportParams
+from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
+from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
+
+load_dotenv(override=True)
+
+BASE_FILENAME = "/tmp/pipecat_conversation_"
+
+
+async def fetch_weather_from_api(params: FunctionCallParams):
+    temperature = 75 if params.arguments["format"] == "fahrenheit" else 24
+    await params.result_callback(
+        {
+            "conditions": "nice",
+            "temperature": temperature,
+            "format": params.arguments["format"],
+            "timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"),
+        }
+    )
+
+
+async def get_saved_conversation_filenames(params: FunctionCallParams):
+    # Construct the full pattern including the BASE_FILENAME
+    full_pattern = f"{BASE_FILENAME}*.json"
+
+    # Use glob to find all matching files
+    matching_files = glob.glob(full_pattern)
+    logger.debug(f"matching files: {matching_files}")
+
+    await params.result_callback({"filenames": matching_files})
+
+
+# async def get_saved_conversation_filenames(
+#     function_name, tool_call_id, args, llm, context, result_callback
+# ):
+#     pattern = re.compile(re.escape(BASE_FILENAME) + "\\d{8}_\\d{6}\\.json$")
+#     matching_files = []
+
+#     for filename in os.listdir("."):
+#         if pattern.match(filename):
+#             matching_files.append(filename)
+
+#     await result_callback({"filenames": matching_files})
+
+
+async def save_conversation(params: FunctionCallParams):
+    timestamp = datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
+    filename = f"{BASE_FILENAME}{timestamp}.json"
+    try:
+        with open(filename, "w") as file:
+            messages = params.context.get_messages_for_persistent_storage()
+            # remove the last few messages. in reverse order, they are:
+            # - the in progress save tool call
+            # - the invocation of the save tool call
+            # - the user ask to save (which may encompass one or more messages)
+            # the simplest thing to do is to pop messages until the last one is an assistant
+            # response
+            while messages and not (
+                messages[-1].get("role") == "assistant" and "content" in messages[-1]
+            ):
+                messages.pop()
+            if messages:  # we never expect this to be empty
+                logger.debug(
+                    f"writing conversation to {filename}\n{json.dumps(messages, indent=4)}"
+                )
+                json.dump(messages, file, indent=2)
+        await params.result_callback({"success": True})
+    except Exception as e:
+        await params.result_callback({"success": False, "error": str(e)})
+
+
+async def load_conversation(params: FunctionCallParams):
+    async def _reset():
+        filename = params.arguments["filename"]
+        logger.debug(f"loading conversation from {filename}")
+        try:
+            with open(filename, "r") as file:
+                messages = json.load(file)
+                messages.append(
+                    {
+                        "role": "user",
+                        "content": f"{AWSNovaSonicLLMService.AWAIT_TRIGGER_ASSISTANT_RESPONSE_INSTRUCTION}",
+                    }
+                )
+                params.context.set_messages(messages)
+                await params.llm.reset_conversation()
+                await params.llm.trigger_assistant_response()
+        except Exception as e:
+            await params.result_callback({"success": False, "error": str(e)})
+
+    asyncio.create_task(_reset())
+
+
+get_current_weather_tool = FunctionSchema(
+    name="get_current_weather",
+    description="Get the current weather",
+    properties={
+        "location": {
+            "type": "string",
+            "description": "The city and state, e.g. San Francisco, CA",
+        },
+        "format": {
+            "type": "string",
+            "enum": ["celsius", "fahrenheit"],
+            "description": "The temperature unit to use. Infer this from the user's location.",
+        },
+    },
+    required=["location", "format"],
+)
+
+save_conversation_tool = FunctionSchema(
+    name="save_conversation",
+    description="Save the current conversation. Use this function to persist the current conversation to external storage.",
+    properties={},
+    required=[],
+)
+
+get_saved_conversation_filenames_tool = FunctionSchema(
+    name="get_saved_conversation_filenames",
+    description="Get a list of saved conversation histories. Returns a list of filenames. Each filename includes a date and timestamp. Each file is conversation history that can be loaded into this session.",
+    properties={},
+    required=[],
+)
+
+load_conversation_tool = FunctionSchema(
+    name="load_conversation",
+    description="Load a conversation history. Use this function to load a conversation history into the current session.",
+    properties={
+        "filename": {
+            "type": "string",
+            "description": "The filename of the conversation history to load.",
+        }
+    },
+    required=["filename"],
+)
+
+tools = ToolsSchema(
+    standard_tools=[
+        get_current_weather_tool,
+        save_conversation_tool,
+        get_saved_conversation_filenames_tool,
+        load_conversation_tool,
+    ]
+)
+
+
+async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):
+    logger.info(f"Starting bot")
+
+    transport = SmallWebRTCTransport(
+        webrtc_connection=webrtc_connection,
+        params=TransportParams(
+            audio_in_enabled=True,
+            audio_out_enabled=True,
+            vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.8)),
+        ),
+    )
+
+    # 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)
+    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. "
+        f"{AWSNovaSonicLLMService.AWAIT_TRIGGER_ASSISTANT_RESPONSE_INSTRUCTION}"
+    )
+
+    llm = AWSNovaSonicLLMService(
+        secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
+        access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
+        region=os.getenv("AWS_REGION"),  # as of 2025-05-06, us-east-1 is the only supported region
+        voice_id="tiffany",  # matthew, tiffany, amy
+        # you could choose to pass instruction here rather than via context
+        # system_instruction=system_instruction,
+        # you could choose to pass tools here rather than via context
+        # tools=tools
+    )
+
+    llm.register_function("get_current_weather", fetch_weather_from_api)
+    llm.register_function("save_conversation", save_conversation)
+    llm.register_function("get_saved_conversation_filenames", get_saved_conversation_filenames)
+    llm.register_function("load_conversation", load_conversation)
+
+    context = OpenAILLMContext(
+        messages=[
+            {"role": "system", "content": f"{system_instruction}"},
+        ],
+        tools=tools,
+    )
+    context_aggregator = llm.create_context_aggregator(context)
+
+    pipeline = Pipeline(
+        [
+            transport.input(),  # Transport user input
+            context_aggregator.user(),
+            llm,  # LLM
+            transport.output(),  # Transport bot output
+            context_aggregator.assistant(),
+        ]
+    )
+
+    task = PipelineTask(
+        pipeline,
+        params=PipelineParams(
+            allow_interruptions=True,
+            enable_metrics=True,
+            enable_usage_metrics=True,
+            report_only_initial_ttfb=True,
+        ),
+    )
+
+    @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([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()
+
+    @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()
+
+    runner = PipelineRunner(handle_sigint=False)
+
+    await runner.run(task)
+
+
+if __name__ == "__main__":
+    from run import main
+
+    main()

examples/foundational/39-aws-nova-sonic.py

@@ -0,0 +1,173 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import argparse
+import os
+from datetime import datetime
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.adapters.schemas.function_schema import FunctionSchema
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.audio.vad.vad_analyzer import VADParams
+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.aws_nova_sonic import AWSNovaSonicLLMService
+from pipecat.services.llm_service import FunctionCallParams
+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 fetch_weather_from_api(params: FunctionCallParams):
+    temperature = 75 if params.arguments["format"] == "fahrenheit" else 24
+    await params.result_callback(
+        {
+            "conditions": "nice",
+            "temperature": temperature,
+            "format": params.arguments["format"],
+            "timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"),
+        }
+    )
+
+
+weather_function = FunctionSchema(
+    name="get_current_weather",
+    description="Get the current weather",
+    properties={
+        "location": {
+            "type": "string",
+            "description": "The city and state, e.g. San Francisco, CA",
+        },
+        "format": {
+            "type": "string",
+            "enum": ["celsius", "fahrenheit"],
+            "description": "The temperature unit to use. Infer this from the users location.",
+        },
+    },
+    required=["location", "format"],
+)
+
+# Create tools schema
+tools = ToolsSchema(standard_tools=[weather_function])
+
+
+async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace):
+    logger.info(f"Starting bot")
+
+    # Initialize the SmallWebRTCTransport with the connection
+    transport = SmallWebRTCTransport(
+        webrtc_connection=webrtc_connection,
+        params=TransportParams(
+            audio_in_enabled=True,
+            audio_in_sample_rate=16000,
+            audio_out_enabled=True,
+            camera_in_enabled=False,
+            vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.8)),
+        ),
+    )
+
+    # 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)
+    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. "
+        f"{AWSNovaSonicLLMService.AWAIT_TRIGGER_ASSISTANT_RESPONSE_INSTRUCTION}"
+    )
+
+    # Create the AWS Nova Sonic LLM service
+    llm = AWSNovaSonicLLMService(
+        secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
+        access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
+        region=os.getenv("AWS_REGION"),  # as of 2025-05-06, us-east-1 is the only supported region
+        voice_id="tiffany",  # matthew, tiffany, amy
+        # you could choose to pass instruction here rather than via context
+        # system_instruction=system_instruction
+        # you could choose to pass tools here rather than via context
+        # tools=tools
+    )
+
+    # Register function for function calls
+    # you can either register a single function for all function calls, or specific functions
+    # llm.register_function(None, fetch_weather_from_api)
+    llm.register_function("get_current_weather", fetch_weather_from_api)
+
+    # Set up context and context management.
+    # AWSNovaSonicService will adapt OpenAI LLM context objects with standard message format to
+    # what's expected by Nova Sonic.
+    context = OpenAILLMContext(
+        messages=[
+            {"role": "system", "content": f"{system_instruction}"},
+            {
+                "role": "user",
+                "content": "Tell me a fun fact!",
+            },
+        ],
+        tools=tools,
+    )
+    context_aggregator = llm.create_context_aggregator(context)
+
+    # Build the pipeline
+    pipeline = Pipeline(
+        [
+            transport.input(),
+            context_aggregator.user(),
+            llm,
+            transport.output(),
+            context_aggregator.assistant(),
+        ]
+    )
+
+    # 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([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")
+    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()

pyproject.toml

@@ -42,6 +42,7 @@ Website = "https://pipecat.ai"
 anthropic = [ "anthropic~=0.49.0" ]
 assemblyai = [ "assemblyai~=0.37.0" ]
 aws = [ "boto3~=1.37.16", "websockets~=13.1" ]
+aws-nova-sonic = [ "aws_sdk_bedrock_runtime~=0.0.2" ]
 azure = [ "azure-cognitiveservices-speech~=1.42.0"]
 cartesia = [ "cartesia~=1.4.0", "websockets~=13.1" ]
 cerebras = []
@@ -96,6 +97,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"

src/pipecat/adapters/services/aws_nova_sonic_adapter.py

@@ -0,0 +1,40 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+import json
+from typing import Any, Dict, List
+
+from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
+from pipecat.adapters.schemas.function_schema import FunctionSchema
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+
+
+class AWSNovaSonicLLMAdapter(BaseLLMAdapter):
+    @staticmethod
+    def _to_aws_nova_sonic_function_format(function: FunctionSchema) -> Dict[str, Any]:
+        return {
+            "toolSpec": {
+                "name": function.name,
+                "description": function.description,
+                "inputSchema": {
+                    "json": json.dumps(
+                        {
+                            "type": "object",
+                            "properties": function.properties,
+                            "required": function.required,
+                        }
+                    )
+                },
+            }
+        }
+
+    def to_provider_tools_format(self, tools_schema: ToolsSchema) -> List[Dict[str, Any]]:
+        """Converts function schemas to AWS Nova Sonic function-calling format.
+
+        :return: AWS Nova Sonic formatted function call definition.
+        """
+
+        functions_schema = tools_schema.standard_tools
+        return [self._to_aws_nova_sonic_function_format(func) for func in functions_schema]

src/pipecat/services/aws_nova_sonic/__init__.py

@@ -0,0 +1 @@
+from .aws import AWSNovaSonicLLMService

src/pipecat/services/aws_nova_sonic/aws.py

@@ -0,0 +1,975 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import base64
+import json
+import time
+import uuid
+import wave
+from dataclasses import dataclass
+from enum import Enum
+from importlib.resources import files
+from typing import Any, List, Optional
+
+from loguru import logger
+from pydantic import BaseModel, Field
+
+from pipecat.adapters.schemas.tools_schema import ToolsSchema
+from pipecat.adapters.services.aws_nova_sonic_adapter import AWSNovaSonicLLMAdapter
+from pipecat.frames.frames import (
+    BotStoppedSpeakingFrame,
+    CancelFrame,
+    EndFrame,
+    Frame,
+    InputAudioRawFrame,
+    LLMFullResponseEndFrame,
+    LLMFullResponseStartFrame,
+    LLMTextFrame,
+    StartFrame,
+    TranscriptionFrame,
+    TTSAudioRawFrame,
+    TTSStartedFrame,
+    TTSStoppedFrame,
+    TTSTextFrame,
+)
+from pipecat.processors.aggregators.llm_response import (
+    LLMAssistantAggregatorParams,
+    LLMUserAggregatorParams,
+)
+from pipecat.processors.aggregators.openai_llm_context import (
+    OpenAILLMContext,
+    OpenAILLMContextFrame,
+)
+from pipecat.processors.frame_processor import FrameDirection
+from pipecat.services.aws_nova_sonic.context import (
+    AWSNovaSonicAssistantContextAggregator,
+    AWSNovaSonicContextAggregatorPair,
+    AWSNovaSonicLLMContext,
+    AWSNovaSonicUserContextAggregator,
+    Role,
+)
+from pipecat.services.aws_nova_sonic.frames import AWSNovaSonicFunctionCallResultFrame
+from pipecat.services.llm_service import LLMService
+from pipecat.utils.time import time_now_iso8601
+
+try:
+    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
+except ModuleNotFoundError as e:
+    logger.error(f"Exception: {e}")
+    logger.error(
+        "In order to use AWS services, you need to `pip install pipecat-ai[aws-nova-sonic]`."
+    )
+    raise Exception(f"Missing module: {e}")
+
+
+class AWSNovaSonicUnhandledFunctionException(Exception):
+    pass
+
+
+class ContentType(Enum):
+    AUDIO = "AUDIO"
+    TEXT = "TEXT"
+    TOOL = "TOOL"
+
+
+class TextStage(Enum):
+    FINAL = "FINAL"  # what has been said
+    SPECULATIVE = "SPECULATIVE"  # what's planned to be said
+
+
+@dataclass
+class CurrentContent:
+    type: ContentType
+    role: Role
+    text_stage: TextStage  # None if not text
+    text_content: str  # starts as None, then fills in if text
+
+    def __str__(self):
+        return (
+            f"CurrentContent(\n"
+            f"  type={self.type.name},\n"
+            f"  role={self.role.name},\n"
+            f"  text_stage={self.text_stage.name if self.text_stage else 'None'}\n"
+            f")"
+        )
+
+
+class Params(BaseModel):
+    # Audio input
+    input_sample_rate: Optional[int] = Field(default=16000)
+    input_sample_size: Optional[int] = Field(default=16)
+    input_channel_count: Optional[int] = Field(default=1)
+
+    # Audio output
+    output_sample_rate: Optional[int] = Field(default=24000)
+    output_sample_size: Optional[int] = Field(default=16)
+    output_channel_count: Optional[int] = Field(default=1)
+
+    # Inference
+    max_tokens: Optional[int] = Field(default=1024)
+    top_p: Optional[float] = Field(default=0.9)
+    temperature: Optional[float] = Field(default=0.7)
+
+
+class AWSNovaSonicLLMService(LLMService):
+    # Override the default adapter to use the AWSNovaSonicLLMAdapter one
+    adapter_class = AWSNovaSonicLLMAdapter
+
+    def __init__(
+        self,
+        *,
+        secret_access_key: str,
+        access_key_id: str,
+        region: str,
+        model: str = "amazon.nova-sonic-v1:0",
+        voice_id: str = "matthew",  # matthew, tiffany, amy
+        params: Params = Params(),
+        system_instruction: Optional[str] = None,
+        tools: Optional[ToolsSchema] = None,
+        send_transcription_frames: bool = True,
+        **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: Optional[BedrockRuntimeClient] = None
+        self._voice_id = voice_id
+        self._params = params
+        self._system_instruction = system_instruction
+        self._tools = tools
+        self._send_transcription_frames = send_transcription_frames
+        self._context: Optional[AWSNovaSonicLLMContext] = None
+        self._stream: Optional[
+            DuplexEventStream[
+                InvokeModelWithBidirectionalStreamInput,
+                InvokeModelWithBidirectionalStreamOutput,
+                InvokeModelWithBidirectionalStreamOperationOutput,
+            ]
+        ] = None
+        self._receive_task: Optional[asyncio.Task] = None
+        self._prompt_name: Optional[str] = None
+        self._input_audio_content_name: Optional[str] = None
+        self._content_being_received: Optional[CurrentContent] = None
+        self._assistant_is_responding = False
+        self._ready_to_send_context = False
+        self._triggering_assistant_response = False
+        self._assistant_response_trigger_audio: Optional[bytes] = (
+            None  # Not cleared on _disconnect()
+        )
+        self._disconnecting = False
+        self._connected_time: Optional[float] = None
+        self._wants_connection = False
+
+    #
+    # standard AIService frame handling
+    #
+
+    async def start(self, frame: StartFrame):
+        await super().start(frame)
+        self._wants_connection = True
+        await self._start_connecting()
+
+    async def stop(self, frame: EndFrame):
+        await super().stop(frame)
+        self._wants_connection = False
+        await self._disconnect()
+
+    async def cancel(self, frame: CancelFrame):
+        await super().cancel(frame)
+        self._wants_connection = False
+        await self._disconnect()
+
+    #
+    # conversation resetting
+    #
+
+    async def reset_conversation(self):
+        logger.debug("Resetting conversation")
+        await self._handle_bot_stopped_speaking()
+
+        # Carry over previous context through disconnect
+        context = self._context
+        await self._disconnect()
+        self._context = context
+
+        await self._start_connecting()
+
+    #
+    # frame processing
+    #
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, OpenAILLMContextFrame):
+            await self._handle_context(frame.context)
+        elif isinstance(frame, InputAudioRawFrame):
+            await self._handle_input_audio_frame(frame)
+        elif isinstance(frame, BotStoppedSpeakingFrame):
+            await self._handle_bot_stopped_speaking()
+        elif isinstance(frame, AWSNovaSonicFunctionCallResultFrame):
+            await self._handle_function_call_result(frame)
+
+        await self.push_frame(frame, direction)
+
+    async def _handle_context(self, context: OpenAILLMContext):
+        if not self._context:
+            # We got our initial context - try to finish connecting
+            self._context = AWSNovaSonicLLMContext.upgrade_to_nova_sonic(
+                context, self._system_instruction
+            )
+            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
+
+        await self._send_user_audio_event(frame.audio)
+
+    async def _handle_bot_stopped_speaking(self):
+        if self._assistant_is_responding:
+            # Consider the assistant finished with their response (after a short delay, to allow for
+            # any FINAL text block to come in).
+            #
+            # TODO: ideally we could base this solely on the LLM output events, but I couldn't
+            # figure out a reliable way to determine when we've gotten our last FINAL text block
+            # after the LLM is done talking.
+            #
+            # First I looked at stopReason, but it doesn't seem like the last FINAL text block is
+            # reliably marked END_TURN (sometimes the *first* one is, but not the last...bug?)
+            #
+            # Then I considered schemes where we tally or match up SPECULATIVE text blocks with
+            # FINAL text blocks to know how many or which FINAL blocks to expect, but user
+            # interruptions throw a wrench in these schemes: depending on the exact timing of the
+            # interruption, we should or shouldn't expect some FINAL blocks.
+            await asyncio.sleep(0.25)
+            self._assistant_is_responding = False
+            await self._report_assistant_response_ended()
+
+    async def _handle_function_call_result(self, frame: AWSNovaSonicFunctionCallResultFrame):
+        result = frame.result_frame
+        await self._send_tool_result(tool_call_id=result.tool_call_id, result=result.result)
+
+    #
+    # LLM communication: lifecycle
+    #
+
+    async def _start_connecting(self):
+        try:
+            logger.info("Connecting...")
+
+            if self._client:
+                # Here we assume that if we have a client we are connected or connecting
+                return
+
+            # Set IDs for the connection
+            self._prompt_name = str(uuid.uuid4())
+            self._input_audio_content_name = str(uuid.uuid4())
+
+            # Create the client
+            self._client = self._create_client()
+
+            # Start the bidirectional stream
+            self._stream = await self._client.invoke_model_with_bidirectional_stream(
+                InvokeModelWithBidirectionalStreamOperationInput(model_id=self._model)
+            )
+
+            # Send session start event
+            await self._send_session_start_event()
+
+            # Finish connecting
+            self._ready_to_send_context = True
+            await self._finish_connecting_if_context_available()
+        except Exception as e:
+            logger.error(f"{self} initialization error: {e}")
+            self._disconnect()
+
+    async def _finish_connecting_if_context_available(self):
+        # We can only finish connecting once we've gotten our initial context and we're ready to
+        # send it
+        if not (self._context and self._ready_to_send_context):
+            return
+
+        logger.info("Finishing connecting (setting up session)...")
+
+        # Read context
+        history = self._context.get_messages_for_initializing_history()
+
+        # Send prompt start event, specifying tools.
+        # Tools from context take priority over self._tools.
+        tools = (
+            self._context.tools
+            if self._context.tools
+            else self.get_llm_adapter().from_standard_tools(self._tools)
+        )
+        logger.debug(f"Using tools: {tools}")
+        await self._send_prompt_start_event(tools)
+
+        # Send system instruction.
+        # Instruction from context takes priority over self._system_instruction.
+        # (NOTE: this prioritizing occurred automatically behind the scenes: the context was
+        # initialized with self._system_instruction and then updated itself from its messages when
+        # get_messages_for_initializing_history() was called).
+        logger.debug(f"Using system instruction: {history.system_instruction}")
+        if history.system_instruction:
+            await self._send_text_event(text=history.system_instruction, role=Role.SYSTEM)
+
+        # Send conversation history
+        for message in history.messages:
+            await self._send_text_event(text=message.text, role=message.role)
+
+        # Start audio input
+        await self._send_audio_input_start_event()
+
+        # Start receiving events
+        self._receive_task = self.create_task(self._receive_task_handler())
+
+        # Record finished connecting time (must be done before sending assistant response trigger)
+        self._connected_time = time.time()
+
+        logger.info("Finished connecting")
+
+        # If we need to, send assistant response trigger (depends on self._connected_time)
+        if self._triggering_assistant_response:
+            await self._send_assistant_response_trigger()
+            self._triggering_assistant_response = False
+
+    async def _disconnect(self):
+        try:
+            logger.info("Disconnecting...")
+
+            # NOTE: see explanation of HACK, below
+            self._disconnecting = True
+
+            # Clean up client
+            if self._client:
+                await self._send_session_end_events()
+                self._client = None
+
+            # Clean up stream
+            if self._stream:
+                await self._stream.input_stream.close()
+                self._stream = None
+
+            # NOTE: see explanation of HACK, below
+            await asyncio.sleep(1)
+
+            # Clean up receive task
+            # HACK: we should ideally be able to cancel the receive task before stopping the input
+            # stream, above (meaning we wouldn't need self._disconnecting). But for some reason if
+            # we don't close the input stream and wait a second first, we're getting an error a lot
+            # like this one: https://github.com/awslabs/amazon-transcribe-streaming-sdk/issues/61.
+            if self._receive_task:
+                await self.cancel_task(self._receive_task, timeout=1.0)
+                self._receive_task = None
+
+            # Reset remaining connection-specific state
+            self._prompt_name = None
+            self._input_audio_content_name = None
+            self._content_being_received = None
+            self._assistant_is_responding = False
+            self._ready_to_send_context = False
+            self._triggering_assistant_response = False
+            self._disconnecting = False
+            self._connected_time = None
+
+            logger.info("Finished disconnecting")
+        except Exception as e:
+            logger.error(f"{self} error disconnecting: {e}")
+
+    def _create_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
+                )
+            ),
+            http_auth_scheme_resolver=HTTPAuthSchemeResolver(),
+            http_auth_schemes={"aws.auth#sigv4": SigV4AuthScheme()},
+        )
+        return BedrockRuntimeClient(config=config)
+
+    #
+    # LLM communication: input events (pipecat -> LLM)
+    #
+
+    async def _send_session_start_event(self):
+        session_start = f"""
+        {{
+          "event": {{
+            "sessionStart": {{
+              "inferenceConfiguration": {{
+                "maxTokens": {self._params.max_tokens},
+                "topP": {self._params.top_p},
+                "temperature": {self._params.temperature}
+              }}
+            }}
+          }}
+        }}
+        """
+        await self._send_client_event(session_start)
+
+    async def _send_prompt_start_event(self, tools: List[Any]):
+        if not self._prompt_name:
+            return
+
+        tools_config = (
+            f""",
+        "toolUseOutputConfiguration": {{
+          "mediaType": "application/json"
+        }},
+        "toolConfiguration": {{
+          "tools": {json.dumps(tools)}
+        }}
+        """
+            if tools
+            else ""
+        )
+
+        prompt_start = f'''
+        {{
+          "event": {{
+            "promptStart": {{
+              "promptName": "{self._prompt_name}",
+              "textOutputConfiguration": {{
+                "mediaType": "text/plain"
+              }},
+              "audioOutputConfiguration": {{
+                "mediaType": "audio/lpcm",
+                "sampleRateHertz": {self._params.output_sample_rate},
+                "sampleSizeBits": {self._params.output_sample_size},
+                "channelCount": {self._params.output_channel_count},
+                "voiceId": "{self._voice_id}",
+                "encoding": "base64",
+                "audioType": "SPEECH"
+              }}{tools_config}
+            }}
+          }}
+        }}
+        '''
+        await self._send_client_event(prompt_start)
+
+    async def _send_audio_input_start_event(self):
+        if not self._prompt_name:
+            return
+
+        audio_content_start = f'''
+        {{
+            "event": {{
+                "contentStart": {{
+                    "promptName": "{self._prompt_name}",
+                    "contentName": "{self._input_audio_content_name}",
+                    "type": "AUDIO",
+                    "interactive": true,
+                    "role": "USER",
+                    "audioInputConfiguration": {{
+                        "mediaType": "audio/lpcm",
+                        "sampleRateHertz": {self._params.input_sample_rate},
+                        "sampleSizeBits": {self._params.input_sample_size},
+                        "channelCount": {self._params.input_channel_count},
+                        "audioType": "SPEECH",
+                        "encoding": "base64"
+                    }}
+                }}
+            }}
+        }}
+        '''
+        await self._send_client_event(audio_content_start)
+
+    async def _send_text_event(self, text: str, role: Role):
+        if not self._stream or not self._prompt_name or not text:
+            return
+
+        content_name = str(uuid.uuid4())
+
+        text_content_start = f'''
+        {{
+            "event": {{
+                "contentStart": {{
+                    "promptName": "{self._prompt_name}",
+                    "contentName": "{content_name}",
+                    "type": "TEXT",
+                    "interactive": true,
+                    "role": "{role.value}",
+                    "textInputConfiguration": {{
+                        "mediaType": "text/plain"
+                    }}
+                }}
+            }}
+        }}
+        '''
+        await self._send_client_event(text_content_start)
+
+        escaped_text = json.dumps(text)  # includes quotes
+        text_input = f'''
+        {{
+            "event": {{
+                "textInput": {{
+                    "promptName": "{self._prompt_name}",
+                    "contentName": "{content_name}",
+                    "content": {escaped_text}
+                }}
+            }}
+        }}
+        '''
+        await self._send_client_event(text_input)
+
+        text_content_end = f'''
+        {{
+            "event": {{
+                "contentEnd": {{
+                    "promptName": "{self._prompt_name}",
+                    "contentName": "{content_name}"
+                }}
+            }}
+        }}
+        '''
+        await self._send_client_event(text_content_end)
+
+    async def _send_user_audio_event(self, audio: bytes):
+        if not self._stream:
+            return
+
+        blob = base64.b64encode(audio)
+        audio_event = f'''
+        {{
+            "event": {{
+                "audioInput": {{
+                    "promptName": "{self._prompt_name}",
+                    "contentName": "{self._input_audio_content_name}",
+                    "content": "{blob.decode("utf-8")}"
+                }}
+            }}
+        }}
+        '''
+        await self._send_client_event(audio_event)
+
+    async def _send_session_end_events(self):
+        if not self._stream or not self._prompt_name:
+            return
+
+        prompt_end = f'''
+        {{
+            "event": {{
+                "promptEnd": {{
+                    "promptName": "{self._prompt_name}"
+                }}
+            }}
+        }}
+        '''
+        await self._send_client_event(prompt_end)
+
+        session_end = """
+        {
+            "event": {
+                "sessionEnd": {}
+            }
+        }
+        """
+        await self._send_client_event(session_end)
+
+    async def _send_tool_result(self, tool_call_id, result):
+        if not self._stream or not self._prompt_name:
+            return
+
+        content_name = str(uuid.uuid4())
+
+        result_content_start = f'''
+        {{
+            "event": {{
+                "contentStart": {{
+                    "promptName": "{self._prompt_name}",
+                    "contentName": "{content_name}",
+                    "interactive": false,
+                    "type": "TOOL",
+                    "role": "TOOL",
+                    "toolResultInputConfiguration": {{
+                        "toolUseId": "{tool_call_id}",
+                        "type": "TEXT",
+                        "textInputConfiguration": {{
+                            "mediaType": "text/plain"
+                        }}
+                    }}
+                }}
+            }}
+        }}
+        '''
+        await self._send_client_event(result_content_start)
+
+        result_content = json.dumps(
+            {
+                "event": {
+                    "toolResult": {
+                        "promptName": self._prompt_name,
+                        "contentName": content_name,
+                        "content": json.dumps(result) if isinstance(result, dict) else result,
+                    }
+                }
+            }
+        )
+        await self._send_client_event(result_content)
+
+        result_content_end = f"""
+        {{
+            "event": {{
+                "contentEnd": {{
+                    "promptName": "{self._prompt_name}",
+                    "contentName": "{content_name}"
+                }}
+            }}
+        }}
+        """
+        await self._send_client_event(result_content_end)
+
+    async def _send_client_event(self, event_json: str):
+        if not self._stream:  # should never happen
+            return
+
+        event = InvokeModelWithBidirectionalStreamInputChunk(
+            value=BidirectionalInputPayloadPart(bytes_=event_json.encode("utf-8"))
+        )
+        await self._stream.input_stream.send(event)
+
+    #
+    # LLM communication: output events (LLM -> pipecat)
+    #
+
+    # Receive events for the session.
+    # A few different kinds of content can be delivered:
+    # - Transcription of user audio
+    # - Tool use
+    # - Text preview of planned response speech before audio delivered
+    # - User interruption notification
+    # - Text of response speech that whose audio was actually delivered
+    # - Audio of response speech
+    # Each piece of content is wrapped by "contentStart" and "contentEnd" events. The content is
+    # delivered sequentially: one piece of content will end before another starts.
+    # The overall completion is wrapped by "completionStart" and "completionEnd" events.
+    async def _receive_task_handler(self):
+        try:
+            while self._stream and not self._disconnecting:
+                output = await self._stream.await_output()
+                result = await output[1].receive()
+
+                if result.value and result.value.bytes_:
+                    response_data = result.value.bytes_.decode("utf-8")
+                    json_data = json.loads(response_data)
+
+                    if "event" in json_data:
+                        event_json = json_data["event"]
+                        if "completionStart" in event_json:
+                            # Handle the LLM completion 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 "toolUse" in event_json:
+                            # Handle tool use
+                            await self._handle_tool_use_event(event_json)
+                        elif "contentEnd" in event_json:
+                            # Handle a piece of content ending
+                            await self._handle_content_end_event(event_json)
+                        elif "completionEnd" in event_json:
+                            # Handle the LLM completion ending
+                            await self._handle_completion_end_event(event_json)
+
+        except Exception as e:
+            logger.error(f"{self} error processing responses: {e}")
+            if self._wants_connection:
+                await self.reset_conversation()
+
+    async def _handle_completion_start_event(self, event_json):
+        pass
+
+    async def _handle_content_start_event(self, event_json):
+        content_start = event_json["contentStart"]
+        type = content_start["type"]
+        role = content_start["role"]
+        generation_stage = None
+        if "additionalModelFields" in content_start:
+            additional_model_fields = json.loads(content_start["additionalModelFields"])
+            generation_stage = additional_model_fields.get("generationStage")
+
+        # Bookkeeping: track current content being received
+        content = CurrentContent(
+            type=ContentType(type),
+            role=Role(role),
+            text_stage=TextStage(generation_stage) if generation_stage else None,
+            text_content=None,
+        )
+        self._content_being_received = content
+
+        if content.role == Role.ASSISTANT:
+            if content.type == ContentType.AUDIO:
+                # Note that an assistant response can comprise of multiple audio blocks
+                if not self._assistant_is_responding:
+                    # The assistant has started responding.
+                    self._assistant_is_responding = True
+                    await self._report_assistant_response_started()
+
+    async def _handle_text_output_event(self, event_json):
+        if not self._content_being_received:  # should never happen
+            return
+        content = self._content_being_received
+
+        text_content = event_json["textOutput"]["content"]
+
+        # Bookkeeping: augment the current content being received with text
+        # Assumption: only one text content per content block
+        content.text_content = text_content
+
+    async def _handle_audio_output_event(self, event_json):
+        if not self._content_being_received:  # should never happen
+            return
+
+        # Get audio
+        audio_content = event_json["audioOutput"]["content"]
+
+        # Push audio frame
+        audio = base64.b64decode(audio_content)
+        frame = TTSAudioRawFrame(
+            audio=audio,
+            sample_rate=self._params.output_sample_rate,
+            num_channels=self._params.output_channel_count,
+        )
+        await self.push_frame(frame)
+
+    async def _handle_tool_use_event(self, event_json):
+        if not self._content_being_received or not self._context:  # should never happen
+            return
+
+        # Get tool use details
+        tool_use = event_json["toolUse"]
+        function_name = tool_use["toolName"]
+        tool_call_id = tool_use["toolUseId"]
+        arguments = json.loads(tool_use["content"])
+
+        # Call tool function
+        if self.has_function(function_name):
+            if function_name in self._functions.keys() or None in self._functions.keys():
+                await self.call_function(
+                    context=self._context,
+                    tool_call_id=tool_call_id,
+                    function_name=function_name,
+                    arguments=arguments,
+                )
+        else:
+            raise AWSNovaSonicUnhandledFunctionException(
+                f"The LLM tried to call a function named '{function_name}', but there isn't a callback registered for that function."
+            )
+
+    async def _handle_content_end_event(self, event_json):
+        if not self._content_being_received:  # should never happen
+            return
+        content = self._content_being_received
+
+        content_end = event_json["contentEnd"]
+        stop_reason = content_end["stopReason"]
+
+        # Bookkeeping: clear current content being received
+        self._content_being_received = None
+
+        if content.role == Role.ASSISTANT:
+            if content.type == ContentType.TEXT:
+                # Ignore non-final text, and the "interrupted" message (which isn't meaningful text)
+                if content.text_stage == TextStage.FINAL and stop_reason != "INTERRUPTED":
+                    if self._assistant_is_responding:
+                        # Text added to the ongoing assistant response
+                        await self._report_assistant_response_text_added(content.text_content)
+        elif content.role == Role.USER:
+            if content.type == ContentType.TEXT:
+                if content.text_stage == TextStage.FINAL:
+                    # User transcription text added
+                    await self._report_user_transcription_text_added(content.text_content)
+
+    async def _handle_completion_end_event(self, event_json):
+        pass
+
+    async def _report_assistant_response_started(self):
+        logger.debug("Assistant response started")
+
+        # Report that the assistant has started their response.
+        await self.push_frame(LLMFullResponseStartFrame())
+
+        # Report that equivalent of TTS (this is a speech-to-speech model) started
+        await self.push_frame(TTSStartedFrame())
+
+    async def _report_assistant_response_text_added(self, text):
+        if not self._context:  # should never happen
+            return
+
+        logger.debug(f"Assistant response text added: {text}")
+
+        # Report some text added to the ongoing assistant response
+        await self.push_frame(LLMTextFrame(text))
+
+        # Report some text added to the *equivalent* of TTS (this is a speech-to-speech model)
+        await self.push_frame(TTSTextFrame(text))
+
+        # TODO: this is a (hopefully temporary) HACK. Here we directly manipulate the context rather
+        # than relying on the frames pushed to the assistant context aggregator. The pattern of
+        # receiving full-sentence text after the assistant has spoken does not easily fit with the
+        # Pipecat expectation of chunks of text streaming in while the assistant is speaking.
+        # Interruption handling was especially challenging. Rather than spend days trying to fit a
+        # square peg in a round hole, I decided on this hack for the time being. We can most cleanly
+        # abandon this hack if/when AWS Nova Sonic implements streaming smaller text chunks
+        # interspersed with audio. Note that when we move away from this hack, we need to make sure
+        # that on an interruption we avoid sending LLMFullResponseEndFrame, which gets the
+        # LLMAssistantContextAggregator into a bad state.
+        self._context.buffer_assistant_text(text)
+
+    async def _report_assistant_response_ended(self):
+        if not self._context:  # should never happen
+            return
+
+        logger.debug("Assistant response ended")
+
+        # Report that the assistant has finished their response.
+        await self.push_frame(LLMFullResponseEndFrame())
+
+        # Report that equivalent of TTS (this is a speech-to-speech model) stopped.
+        await self.push_frame(TTSStoppedFrame())
+
+        # For an explanation of this hack, see _report_assistant_response_text_added.
+        self._context.flush_aggregated_assistant_text()
+
+    async def _report_user_transcription_text_added(self, text):
+        if not self._context:  # should never happen
+            return
+
+        logger.debug(f"User transcription text added: {text}")
+
+        # Manually add new user transcription text to context.
+        # We can't rely on the user context aggregator to do this since it's upstream from the LLM.
+        self._context.add_user_transcription_text(text)
+
+        # Report that some new user transcription text is available.
+        if self._send_transcription_frames:
+            await self.push_frame(
+                TranscriptionFrame(text=text, user_id="", timestamp=time_now_iso8601())
+            )
+
+    #
+    # context
+    #
+
+    def create_context_aggregator(
+        self,
+        context: OpenAILLMContext,
+        *,
+        user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
+        assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
+    ) -> AWSNovaSonicContextAggregatorPair:
+        context.set_llm_adapter(self.get_llm_adapter())
+
+        user = AWSNovaSonicUserContextAggregator(context=context, params=user_params)
+        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
+        if self._connected_time is not None:
+            await self._send_assistant_response_trigger()
+            self._triggering_assistant_response = False
+
+    async def _send_assistant_response_trigger(self):
+        if (
+            not self._assistant_response_trigger_audio or self._connected_time is None
+        ):  # should never happen
+            return
+
+        logger.debug("Sending assistant response trigger...")
+
+        chunk_duration = 0.02  # what we might get from InputAudioRawFrame
+        chunk_size = int(
+            chunk_duration
+            * self._params.input_sample_rate
+            * self._params.input_channel_count
+            * (self._params.input_sample_size / 8)
+        )  # e.g. 0.02 seconds of 16-bit (2-byte) PCM mono audio at 16kHz is 640 bytes
+
+        # Lead with a bit of blank audio, if needed.
+        # It seems like the LLM can't quite "hear" the first little bit of audio sent on a
+        # connection.
+        current_time = time.time()
+        max_blank_audio_duration = 0.5
+        blank_audio_duration = (
+            max_blank_audio_duration - (current_time - self._connected_time)
+            if self._connected_time is not None
+            and (current_time - self._connected_time) < max_blank_audio_duration
+            else None
+        )
+        if blank_audio_duration:
+            logger.debug(
+                f"Leading assistant response trigger with {blank_audio_duration}s of blank audio"
+            )
+            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.
+        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)

src/pipecat/services/aws_nova_sonic/context.py

@@ -0,0 +1,217 @@
+#
+# Copyright (c) 2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import copy
+from dataclasses import dataclass, field
+from enum import Enum
+
+from loguru import logger
+
+from pipecat.frames.frames import (
+    BotStoppedSpeakingFrame,
+    DataFrame,
+    Frame,
+    FunctionCallResultFrame,
+    LLMFullResponseEndFrame,
+    LLMFullResponseStartFrame,
+    LLMMessagesAppendFrame,
+    LLMMessagesUpdateFrame,
+    LLMSetToolChoiceFrame,
+    LLMSetToolsFrame,
+    StartInterruptionFrame,
+    TextFrame,
+    UserImageRawFrame,
+)
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.frame_processor import FrameDirection
+from pipecat.services.aws_nova_sonic.frames import AWSNovaSonicFunctionCallResultFrame
+from pipecat.services.openai.llm import (
+    OpenAIAssistantContextAggregator,
+    OpenAIUserContextAggregator,
+)
+
+
+class Role(Enum):
+    SYSTEM = "SYSTEM"
+    USER = "USER"
+    ASSISTANT = "ASSISTANT"
+    TOOL = "TOOL"
+
+
+@dataclass
+class AWSNovaSonicConversationHistoryMessage:
+    role: Role  # only USER and ASSISTANT
+    text: str
+
+
+@dataclass
+class AWSNovaSonicConversationHistory:
+    system_instruction: str = None
+    messages: list[AWSNovaSonicConversationHistoryMessage] = field(default_factory=list)
+
+
+class AWSNovaSonicLLMContext(OpenAILLMContext):
+    def __init__(self, messages=None, tools=None, **kwargs):
+        super().__init__(messages=messages, tools=tools, **kwargs)
+        self.__setup_local()
+
+    def __setup_local(self, system_instruction: str = ""):
+        self._assistant_text = ""
+        self._system_instruction = system_instruction
+
+    @staticmethod
+    def upgrade_to_nova_sonic(
+        obj: OpenAILLMContext, system_instruction: str
+    ) -> "AWSNovaSonicLLMContext":
+        if isinstance(obj, OpenAILLMContext) and not isinstance(obj, AWSNovaSonicLLMContext):
+            obj.__class__ = AWSNovaSonicLLMContext
+            obj.__setup_local(system_instruction)
+        return obj
+
+    # NOTE: this method has the side-effect of updating _system_instruction from messages
+    def get_messages_for_initializing_history(self) -> AWSNovaSonicConversationHistory:
+        history = AWSNovaSonicConversationHistory(system_instruction=self._system_instruction)
+
+        # Bail if there are no messages
+        if not self.messages:
+            return history
+
+        messages = copy.deepcopy(self.messages)
+
+        # If we have a "system" message as our first message, let's pull that out into "instruction"
+        if messages[0].get("role") == "system":
+            system = messages.pop(0)
+            content = system.get("content")
+            if isinstance(content, str):
+                history.system_instruction = content
+            elif isinstance(content, list):
+                history.system_instruction = content[0].get("text")
+            if history.system_instruction:
+                self._system_instruction = history.system_instruction
+
+        # Process remaining messages to fill out conversation history.
+        # Nova Sonic supports "user" and "assistant" messages in history.
+        for message in messages:
+            history_message = self.from_standard_message(message)
+            if history_message:
+                history.messages.append(history_message)
+
+        return history
+
+    def get_messages_for_persistent_storage(self):
+        messages = super().get_messages_for_persistent_storage()
+        # If we have a system instruction and messages doesn't already contain it, add it
+        if self._system_instruction and not (messages and messages[0].get("role") == "system"):
+            messages.insert(0, {"role": "system", "content": self._system_instruction})
+        return messages
+
+    def from_standard_message(self, message) -> AWSNovaSonicConversationHistoryMessage:
+        role = message.get("role")
+        if message.get("role") == "user" or message.get("role") == "assistant":
+            content = message.get("content")
+            if isinstance(message.get("content"), list):
+                content = ""
+                for c in message.get("content"):
+                    if c.get("type") == "text":
+                        content += " " + c.get("text")
+                    else:
+                        logger.error(
+                            f"Unhandled content type in context message: {c.get('type')} - {message}"
+                        )
+            # There won't be content if this is an assistant tool call entry.
+            # We're ignoring those since they can't be loaded into AWS Nova Sonic conversation
+            # history
+            if content:
+                return AWSNovaSonicConversationHistoryMessage(role=Role[role.upper()], text=content)
+        # NOTE: we're ignoring messages with role "tool" since they can't be loaded into AWS Nova
+        # Sonic conversation history
+
+    def add_user_transcription_text(self, text):
+        message = {
+            "role": "user",
+            "content": [{"type": "text", "text": text}],
+        }
+        self.add_message(message)
+        # logger.debug(f"Context updated (user): {self.get_messages_for_logging()}")
+
+    def buffer_assistant_text(self, text):
+        self._assistant_text += text
+        # logger.debug(f"Assistant text buffered: {self._assistant_text}")
+
+    def flush_aggregated_assistant_text(self):
+        if not self._assistant_text:
+            return
+        message = {
+            "role": "assistant",
+            "content": [{"type": "text", "text": self._assistant_text}],
+        }
+        self._assistant_text = ""
+        self.add_message(message)
+        # logger.debug(f"Context updated (assistant): {self.get_messages_for_logging()}")
+
+
+@dataclass
+class AWSNovaSonicMessagesUpdateFrame(DataFrame):
+    context: AWSNovaSonicLLMContext
+
+
+class AWSNovaSonicUserContextAggregator(OpenAIUserContextAggregator):
+    async def process_frame(
+        self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM
+    ):
+        await super().process_frame(frame, direction)
+
+        # Parent does not push LLMMessagesUpdateFrame
+        if isinstance(frame, LLMMessagesUpdateFrame):
+            await self.push_frame(AWSNovaSonicMessagesUpdateFrame(context=self._context))
+
+
+class AWSNovaSonicAssistantContextAggregator(OpenAIAssistantContextAggregator):
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        # HACK: For now, disable the context aggregator by making it just pass through all frames
+        # that the parent handles (except the function call stuff, which we still need).
+        # For an explanation of this hack, see
+        # AWSNovaSonicLLMService._report_assistant_response_text_added.
+        if isinstance(
+            frame,
+            (
+                StartInterruptionFrame,
+                LLMFullResponseStartFrame,
+                LLMFullResponseEndFrame,
+                TextFrame,
+                LLMMessagesAppendFrame,
+                LLMMessagesUpdateFrame,
+                LLMSetToolsFrame,
+                LLMSetToolChoiceFrame,
+                UserImageRawFrame,
+                BotStoppedSpeakingFrame,
+            ),
+        ):
+            await self.push_frame(frame, direction)
+        else:
+            await super().process_frame(frame, direction)
+
+    async def handle_function_call_result(self, frame: FunctionCallResultFrame):
+        await super().handle_function_call_result(frame)
+
+        # The standard function callback code path pushes the FunctionCallResultFrame from the LLM
+        # itself, so we didn't have a chance to add the result to the AWS Nova Sonic server-side
+        # context. Let's push a special frame to do that.
+        await self.push_frame(
+            AWSNovaSonicFunctionCallResultFrame(result_frame=frame), FrameDirection.UPSTREAM
+        )
+
+
+@dataclass
+class AWSNovaSonicContextAggregatorPair:
+    _user: AWSNovaSonicUserContextAggregator
+    _assistant: AWSNovaSonicAssistantContextAggregator
+
+    def user(self) -> AWSNovaSonicUserContextAggregator:
+        return self._user
+
+    def assistant(self) -> AWSNovaSonicAssistantContextAggregator:
+        return self._assistant

src/pipecat/services/aws_nova_sonic/frames.py

@@ -0,0 +1,14 @@
+#
+# Copyright (c) 2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+from dataclasses import dataclass
+
+from pipecat.frames.frames import DataFrame, FunctionCallResultFrame
+
+
+@dataclass
+class AWSNovaSonicFunctionCallResultFrame(DataFrame):
+    result_frame: FunctionCallResultFrame

src/pipecat/services/openai_realtime_beta/openai.py

@@ -577,15 +577,7 @@ class OpenAIRealtimeBetaLLMService(LLMService):
             arguments = json.loads(item.arguments)
             if self.has_function(function_name):
                 run_llm = index == total_items - 1
-                if function_name in self._functions.keys():
-                    await self.call_function(
-                        context=self._context,
-                        tool_call_id=tool_id,
-                        function_name=function_name,
-                        arguments=arguments,
-                        run_llm=run_llm,
-                    )
-                elif None in self._functions.keys():
+                if function_name in self._functions.keys() or None in self._functions.keys():
                     await self.call_function(
                         context=self._context,
                         tool_call_id=tool_id,