diff --git a/examples/foundational/20e-persistent-context-aws-nova-sonic.py b/examples/foundational/20e-persistent-context-aws-nova-sonic.py new file mode 100644 index 000000000..8a95f54b9 --- /dev/null +++ b/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() diff --git a/examples/foundational/39-aws-nova-sonic.py b/examples/foundational/39-aws-nova-sonic.py index 07670f75a..c80626962 100644 --- a/examples/foundational/39-aws-nova-sonic.py +++ b/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 ) diff --git a/src/pipecat/services/aws_nova_sonic/aws.py b/src/pipecat/services/aws_nova_sonic/aws.py index 5b69810f3..50b83d3e0 100644 --- a/src/pipecat/services/aws_nova_sonic/aws.py +++ b/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 - # Tools from context take priority over tools from __init__() + # Send prompt start event, specifying tools. + # 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 - # Instruction from context takes priority over instruction from __init__() - instruction = history.instruction if history.instruction else self._instruction - if instruction: - await self._send_text_event(text=instruction, role=Role.SYSTEM) + # 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). + # 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 - # 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) + await self._send_assistant_response_trigger() self._triggering_assistant_response = False async def _disconnect(self): try: - # Clean up receive task - if self._receive_task: - await self.cancel_task(self._receive_task, timeout=1.0) - self._receive_task = None + # NOTE: see explanation of HACK, below + self._disconnecting = True # 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". - # There is generally a single completion per session. - # In a completion, a few different kinds of content can be delivered: + # 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 @@ -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 - if lead_with_blank_audio: - blank_audio_duration = 0.5 # much less than this and it doesn't reliably work + # 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: blank_audio_chunk = b"\x00" * chunk_size num_chunks = int(blank_audio_duration / chunk_duration) for _ in range(num_chunks): @@ -925,7 +967,7 @@ class AWSNovaSonicLLMService(LLMService): # 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 + # since the instruction (i.e. the "wait for the trigger" instruction) will be part of the # context as well. # print(f"[pk] sending trigger audio! {len(self._assistant_response_trigger_audio)}") audio_chunks = [ diff --git a/src/pipecat/services/aws_nova_sonic/context.py b/src/pipecat/services/aws_nova_sonic/context.py index 3fac65a72..b12061e1e 100644 --- a/src/pipecat/services/aws_nova_sonic/context.py +++ b/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":