[WIP] AWS Nova Sonic service - implement ability to persist and load conversations

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

@@ -0,0 +1,256 @@
+#
+# Copyright (c) 2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+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.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(function_name, tool_call_id, args, llm, context, result_callback):
+    temperature = 75 if args["format"] == "fahrenheit" else 24
+    await result_callback(
+        {
+            "conditions": "nice",
+            "temperature": temperature,
+            "format": args["format"],
+            "timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"),
+        }
+    )
+
+
+async def get_saved_conversation_filenames(
+    function_name, tool_call_id, args, llm, context, result_callback
+):
+    # 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 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(function_name, tool_call_id, args, llm, context, result_callback):
+    timestamp = datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
+    filename = f"{BASE_FILENAME}{timestamp}.json"
+    logger.debug(
+        f"writing conversation to {filename}\n{json.dumps(context.get_messages_for_persistent_storage(), indent=4)}"
+    )
+    try:
+        with open(filename, "w") as file:
+            messages = context.get_messages_for_persistent_storage()
+            # remove the last message, which is the instruction we just gave to save the conversation
+            messages.pop()
+            json.dump(messages, file, indent=2)
+        await result_callback({"success": True})
+    except Exception as e:
+        await result_callback({"success": False, "error": str(e)})
+
+
+async def load_conversation(function_name, tool_call_id, args, llm, context, result_callback):
+    async def _reset():
+        filename = args["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}",
+                    }
+                )
+                context.set_messages(messages)
+                await llm.reset_conversation()
+                await llm.trigger_assistant_response()
+        except Exception as e:
+            await 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):
+    logger.info(f"Starting bot")
+
+    transport = SmallWebRTCTransport(
+        webrtc_connection=webrtc_connection,
+        params=TransportParams(
+            audio_in_enabled=True,
+            audio_out_enabled=True,
+            vad_enabled=True,
+            vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.8)),
+            vad_audio_passthrough=True,
+        ),
+    )
+
+    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"),
+        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

@@ -102,7 +102,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection):
         region=os.getenv("AWS_REGION"),
         voice_id="tiffany",  # matthew, tiffany, amy
         # you could choose to pass instruction here rather than via context
-        # instruction=system_instruction
+        # system_instruction=system_instruction
         # you could choose to pass tools here rather than via context
         # tools=tools
     )

src/pipecat/services/aws_nova_sonic/aws.py

@@ -7,6 +7,7 @@
 import asyncio
 import base64
 import json
+import time
 import uuid
 import wave
 from dataclasses import dataclass
@@ -119,7 +120,7 @@ class AWSNovaSonicLLMService(LLMService):
         region: str,
         model: str = "amazon.nova-sonic-v1:0",
         voice_id: str = "matthew",  # matthew, tiffany, amy
-        instruction: Optional[str] = None,
+        system_instruction: Optional[str] = None,
         tools: Optional[ToolsSchema] = None,
         send_transcription_frames: bool = True,
         **kwargs,
@@ -131,7 +132,7 @@ class AWSNovaSonicLLMService(LLMService):
         self._model = model
         self._client: BedrockRuntimeClient = None
         self._voice_id = voice_id
-        self._instruction = instruction
+        self._system_instruction = system_instruction
         self._tools = tools
         self._send_transcription_frames = send_transcription_frames
         self._context: AWSNovaSonicLLMContext = None
@@ -150,6 +151,8 @@ class AWSNovaSonicLLMService(LLMService):
         self._handling_bot_stopped_speaking = False
         self._triggering_assistant_response = False
         self._assistant_response_trigger_audio: bytes = None  # Not cleared on _disconnect()
+        self._disconnecting = False
+        self._connected_time: float = None
 
     #
     # standard AIService frame handling
@@ -174,6 +177,18 @@ class AWSNovaSonicLLMService(LLMService):
         await super().cancel(frame)
         await self._disconnect()
 
+    #
+    # conversation resetting
+    #
+
+    async def reset_conversation(self):
+        logger.debug("Resetting conversation")
+        await self._disconnect()
+        await self._start_connecting()
+        # Use existing context
+        self._context_available = True
+        await self._finish_connecting_if_context_available()
+
     #
     # frame processing
     #
@@ -207,10 +222,12 @@ class AWSNovaSonicLLMService(LLMService):
 
     async def _handle_context(self, context: OpenAILLMContext):
         # TODO: reset connection if needed (if entirely new context object provided, for instance)
-        print(f"[pk] receive updated context: {context.get_messages_for_initializing_history()}")
+        print(f"[pk] received updated context: {context.get_messages_for_initializing_history()}")
         if not self._context:
             # We got our initial context - try to finish connecting
-            self._context = AWSNovaSonicLLMContext.upgrade_to_nova_sonic(context)
+            self._context = AWSNovaSonicLLMContext.upgrade_to_nova_sonic(
+                context, self._system_instruction
+            )
             self._context_available = True
             await self._finish_connecting_if_context_available()
 
@@ -296,8 +313,8 @@ class AWSNovaSonicLLMService(LLMService):
         # Read context
         history = self._context.get_messages_for_initializing_history()
 
-        # Send prompt start event, specifying tools
+        # Send prompt start event, specifying tools.
-        # Tools from context take priority over tools from __init__()
+        # Tools from context take priority over self._tools.
         tools = (
             self._context.tools
             if self._context.tools
@@ -305,11 +322,14 @@ class AWSNovaSonicLLMService(LLMService):
         )
         await self._send_prompt_start_event(tools)
 
-        # Send system instruction
+        # Send system instruction.
-        # Instruction from context takes priority over instruction from __init__()
+        # Instruction from context takes priority over self._system_instruction.
-        instruction = history.instruction if history.instruction else self._instruction
+        # (NOTE: this prioritizing occurred automatically behind the scenes: the context was
-        if instruction:
+        # initialized with self._system_instruction and then updated itself from its messages when
-            await self._send_text_event(text=instruction, role=Role.SYSTEM)
+        # get_messages_for_initializing_history() was called).
+        # print(f"[pk] connecting, with 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:
@@ -320,7 +340,7 @@ class AWSNovaSonicLLMService(LLMService):
         # - pass additional message(s)
         # - merge init-passed system instruction + context instruction (latter takes precedence)
         # - merge init-passed tools + context tools (latter takes precedence)
-        await self._send_text_event(text=self._instruction, role=Role.SYSTEM)
+        await self._send_text_event(text=self._system_instruction, role=Role.SYSTEM)
 
         # Start audio input
         await self._send_audio_input_start_event()
@@ -328,31 +348,43 @@ class AWSNovaSonicLLMService(LLMService):
         # Start receiving events
         self._receive_task = self.create_task(self._receive_task_handler())
 
-        # If we need to, send assistant response trigger
+        # Record finished connecting time
+        self._connected_time = time.time()
+
+        # If we need to, send assistant response trigger (depends on self._connected_time)
         if self._triggering_assistant_response:
-            # If the trigger was the first audio chunk sent on this connection it'd be ignored (I'm
+            await self._send_assistant_response_trigger()
-            # 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
+            # NOTE: see explanation of HACK, below
-            if self._receive_task:
+            self._disconnecting = True
-                await self.cancel_task(self._receive_task, timeout=1.0)
-                self._receive_task = None
 
             # Clean up client
             if self._client:
+                print("[pk] Cleaning up client")
                 await self._send_session_end_events()
                 self._client = None
 
             # Clean up stream
             if self._stream:
+                print("[pk] Cleaning up 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
@@ -362,6 +394,8 @@ class AWSNovaSonicLLMService(LLMService):
             self._ready_to_send_context = False
             self._handling_bot_stopped_speaking = False
             self._triggering_assistant_response = False
+            self._disconnecting = False
+            self._connected_time = None
         except Exception as e:
             logger.error(f"{self} error disconnecting: {e}")
 
@@ -619,9 +653,8 @@ class AWSNovaSonicLLMService(LLMService):
     # LLM communication: output events (LLM -> pipecat)
     #
 
-    # Receive the ongoing LLM "completion".
+    # Receive events for the session.
-    # There is generally a single completion per session.
+    # A few different kinds of content can be delivered:
-    # In a completion, a few different kinds of content can be delivered:
     # - Transcription of user audio
     # - Tool use
     # - Text preview of planned response speech before audio delivered
@@ -633,7 +666,7 @@ class AWSNovaSonicLLMService(LLMService):
     # The overall completion is wrapped by "completionStart" and "completionEnd" events.
     async def _receive_task_handler(self):
         try:
-            while self._client:
+            while self._client and not self._disconnecting:
                 output = await self._stream.await_output()
                 result = await output[1].receive()
 
@@ -906,16 +939,25 @@ class AWSNovaSonicLLMService(LLMService):
             await self._send_assistant_response_trigger()
             self._triggering_assistant_response = False
 
-    async def _send_assistant_response_trigger(self, lead_with_blank_audio=False):
+    async def _send_assistant_response_trigger(self):
         # 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
+        # Lead with a bit of blank audio, if needed.
-        if lead_with_blank_audio:
+        # It seems like the LLM can't quite "hear" the first little bit of audio sent on a
-            blank_audio_duration = 0.5  # much less than this and it doesn't reliably work
+        # 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:
             blank_audio_chunk = b"\x00" * chunk_size
             num_chunks = int(blank_audio_duration / chunk_duration)
             for _ in range(num_chunks):

src/pipecat/services/aws_nova_sonic/context.py

@@ -49,7 +49,7 @@ class AWSNovaSonicConversationHistoryMessage:
 
 @dataclass
 class AWSNovaSonicConversationHistory:
-    instruction: str = None
+    system_instruction: str = None
     messages: list[AWSNovaSonicConversationHistoryMessage] = field(default_factory=list)
 
 
@@ -58,18 +58,22 @@ class AWSNovaSonicLLMContext(OpenAILLMContext):
         super().__init__(messages=messages, tools=tools, **kwargs)
         self.__setup_local()
 
-    def __setup_local(self):
+    def __setup_local(self, system_instruction: str = ""):
         self._assistant_text = ""
+        self._system_instruction = system_instruction
 
     @staticmethod
-    def upgrade_to_nova_sonic(obj: OpenAILLMContext) -> "AWSNovaSonicLLMContext":
+    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()
+            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()
+        history = AWSNovaSonicConversationHistory(system_instruction=self._system_instruction)
 
         # Bail if there are no messages
         if not self.messages:
@@ -82,13 +86,15 @@ class AWSNovaSonicLLMContext(OpenAILLMContext):
             system = messages.pop(0)
             content = system.get("content")
             if isinstance(content, str):
-                history.instruction = content
+                history.system_instruction = content
             elif isinstance(content, list):
-                history.instruction = content[0].get("text")
+                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.
-        print(f"[pk] standard messages: {messages}")
+        # print(f"[pk] standard messages: {messages}")
         for message in messages:
             history_message = self.from_standard_message(message)
             if history_message:
@@ -96,6 +102,13 @@ class AWSNovaSonicLLMContext(OpenAILLMContext):
 
         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":